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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org