showuon commented on code in PR #13512:
URL: https://github.com/apache/kafka/pull/13512#discussion_r1159579368
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -987,14 +988,6 @@ 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 =
changelogs.get(partition).stateManager.taskId();
- final StreamTask task = (StreamTask) 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
- final long recordsToRestore =
Math.max(changelogMetadata.restoreEndOffset - startOffset, 0L);
- task.initRemainingRecordsToRestore(time, recordsToRestore);
Review Comment:
With these lines, we got some errors related to "cannot cast the task to
`StreamTask`". Might need to investigate why it will happen. Remove them first
to fix the tests.
--
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
