Copilot commented on code in PR #23984:
URL: https://github.com/apache/kafka/pull/23984#discussion_r4393758443
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1461,6 +1461,10 @@ void runOnceWithProcessingThreads() {
taskManager.activeTaskIds(),
taskManager.standbyTaskIds(), commitLatency);
}
}
+
+ // maybeCommit() commits when the uncommitted bytes exceed the
limit, so they only exceed it here when that
+ // commit was skipped (during a rebalance) or failed, or when the
task executors wrote more since the commit
+ maybePauseProcessingForUncommittedBytes();
Review Comment:
With the Streams group protocol, a status-only heartbeat such as
`ASSIGNMENT_DELAYED` can mark the group not ready while retaining active tasks.
The early return at line 1427 then bypasses both commits and this pause check,
so executors can keep processing and punctuating without the byte limit taking
effect. Call `maybePauseProcessingForUncommittedBytes()` before that early
return too, and add a processing-thread test for a ready-to-not-ready
transition with retained tasks.
--
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]
