lucasbru commented on code in PR #14936:
URL: https://github.com/apache/kafka/pull/14936#discussion_r1420236155
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -231,6 +240,69 @@ private void restoreTasks(final long now) {
}
}
+ private void maybeGetClientInstanceIds() {
+ if (fetchDeadlineClientInstanceId != -1) {
+ if (!clientInstanceIdFuture.isDone()) {
+ if (fetchDeadlineClientInstanceId >= time.milliseconds()) {
+ try {
+ // if the state-updated thread has active work:
+ // we pass in a timeout of zero into each
`clientInstanceId()` call
+ // to just trigger the "get instance id"
background RPC;
+ // we don't want to block the state updater
thread that can do useful work in the meantime
+ // otherwise, we pass in 100ms to avoid busy
waiting
+ clientInstanceIdFuture.complete(
+ restoreConsumer.clientInstanceId(
+ allWorkDone() ? Duration.ofMillis(100L) :
Duration.ZERO
Review Comment:
I think this will always be false here, because of
`fetchDeadlineClientInstanceId`
--
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
