mjsax commented on a change in pull request #9368:
URL: https://github.com/apache/kafka/pull/9368#discussion_r499106137
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
##########
@@ -431,6 +432,9 @@ public void restore() {
// in order to make sure we call the main consumer#poll in
time.
// TODO: once we move ChangelogReader to a separate thread
this may no longer be a concern
polledRecords = restoreConsumer.poll(state ==
ChangelogReaderState.STANDBY_UPDATING ? Duration.ZERO : pollTime);
+
+ // TODO (?) If we cannot fetch records during restore, should
we trigger `task.timeout.ms` ?
+ // TODO (?) If we cannot fetch records for standby task,
should we trigger `task.timeout.ms` ?
Review comment:
For the global-thread, we also consider the case that `poll()` returns
nothing. For the global-thread it's a little simper though, as we restore on a
per-partition basis.
For the `StoreChangelogReader` it's more complicated:
- for StandbyTasks, there might be nothing to be restored and thus getting
no records might be fine (however, the "metadata" is not easily available as
it's encapsulated in other classes...
- for active restoring tasks, it might be simpler...
Was holding off, as I am not sure how important this case is, with regard to
robustness. We would not crash for this case. I would still like to get this
done, but not necessarily for `2.7` release though.
----------------------------------------------------------------
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:
us...@infra.apache.org
