mjsax commented on code in PR #13523:
URL: https://github.com/apache/kafka/pull/13523#discussion_r1160999602
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -988,6 +1005,14 @@ private void prepareChangelogs(final Map<TaskId, Task>
tasks,
} catch (final Exception e) {
throw new StreamsException("State restore listener failed
on batch restored", e);
}
+
+ final TaskId taskId = changelogMetadata.stateManager.taskId();
+ final Task task = tasks.get(taskId);
+ // if the log is truncated between when we get the log end
offset and when we get the
+ // consumer position, then it's possible that the difference
become negative and there's actually
+ // no records to restore; in this case we just initialize the
sensor to zero
Review Comment:
Not sure if I can follow? If we get the end-offset first, the end-offset we
have won't change even if the log gets truncated? And the consumer position
should not change anyway? Can you elaborate?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
