sanghyeok An created KAFKA-21301:
------------------------------------
Summary: Streams Group Assignment delays temporarily stop
processing existing Streams tasks
Key: KAFKA-21301
URL: https://issues.apache.org/jira/browse/KAFKA-21301
Project: Kafka
Issue Type: Bug
Reporter: sanghyeok An
Assignee: sanghyeok An
With the KIP-1071 Streams Rebalance Protocol, when the assignment interval
delays a new assignment computation, the coordinator returns
{{ASSIGNMENT_DELAYED}} while retaining the existing assignment.
Upon receiving this status, {{StreamsMembershipManager}} sets
{{isGroupReady=false}} regardless of whether the member already has assigned
tasks. Even when the task assignment is unchanged, the readiness change
triggers an assignment callback, and {{DefaultStreamsRebalanceListener}}
forwards the updated readiness to the StreamThread.
The default StreamThread processing loop returns before processing records when
readiness is false. Consequently, existing tasks that could continue processing
are temporarily stopped while the next assignment computation is deferred.
A possible scenario is:
# Member A has assigned tasks and is processing normally.
# Member B joins, increasing the group epoch and requiring a new assignment
computation.
# The assignment interval has not elapsed since the last computation, so the
coordinator retains the existing target assignment. Member B waits without a
target assignment.
# Member A sends a heartbeat during the same delay window. The coordinator
retains A’s existing assignment but also returns {{{}ASSIGNMENT_DELAYED{}}}.
# Member A sets readiness to false and stops processing its existing tasks.
# Processing resumes after A receives a heartbeat response without the delay
status and completes the assignment callback.
This introduces an unnecessary processing gap despite unchanged task ownership,
potentially increasing record processing latency and reducing throughput. The
default processing loop also skips punctuation and the commit stage for that
iteration.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
