prateekm commented on a change in pull request #1133: SAMZA-2297:
InMemorySystemAdmin offsets are off-by-one in some cases
URL: https://github.com/apache/samza/pull/1133#discussion_r323495628
##########
File path:
samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java
##########
@@ -179,9 +179,10 @@ private SystemStreamMetadata
constructSystemStreamMetadata(
.entrySet()
.stream()
.collect(Collectors.toMap(entry -> entry.getKey().getPartition(),
entry -> {
- String oldestOffset = "0";
- String newestOffset = String.valueOf(entry.getValue().size());
- String upcomingOffset = String.valueOf(entry.getValue().size()
+ 1);
+ List<IncomingMessageEnvelope> messages = entry.getValue();
+ String oldestOffset = messages.isEmpty() ? null : "0";
Review comment:
Seems like it.
https://github.com/apache/samza/blob/master/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java#L403
I need to rely on this information during changelog restore for
transactional state. It doesn't make sense to change the behavior for Kafka as
part of the transactional state changes. Since InMemoryStore is used as a
changelog for tests, I'll relax the assertion/validation for non-null starting
offsets in the restore path. Thanks for the pointer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
