sanghyeok An created KAFKA-21256:
------------------------------------
Summary: Assignment callback failure can stall reconciliation in
the Streams protocol client
Key: KAFKA-21256
URL: https://issues.apache.org/jira/browse/KAFKA-21256
Project: Kafka
Issue Type: Bug
Reporter: sanghyeok An
Assignee: sanghyeok An
When Kafka Streams uses {{{}group.protocol=streams{}}}, a failure in the
{{onTasksAssigned}} callback while processing a new active task assignment can
leave the member unable to make further reconciliation progress.
The consumer’s partition assignment is updated before the callback runs,
whereas {{currentAssignment}} is updated only after the callback succeeds. A
callback failure can therefore leave the actual consumer assignment different
from the last successfully reconciled assignment.
The next reconciliation attempt detects this mismatch and throws an
{{{}IllegalStateException{}}}. However, {{reconciliationInProgress}} has
already been set to {{{}true{}}}, and this exception path does not clear it.
Subsequent attempts return immediately because they assume another
reconciliation is still running.
*Failure scenario*
Consider a member that is successfully processing assignment A and receives a
new assignment A+B:
# Initially, both the consumer assignment and {{currentAssignment}} are A.
# The consumer’s partition assignment is updated to A+B before invoking the
assignment callback.
# The {{onTasksAssigned}} callback throws an exception. The consumer
assignment remains A+B, while {{currentAssignment}} remains A.
# Callback failure handling clears the progress flag, allowing another
reconciliation attempt to begin.
# The next attempt sets the progress flag again, detects the mismatch between
the consumer assignment and {{{}currentAssignment{}}}, and throws an exception.
# The exception leaves the progress flag set. This blocks both retries of the
same target assignment and processing of newer target assignments.
||Stage||Consumer Assignment||Last successufully reconciled assignment
(currentAssignment)||reconciliationInProgress||
|Initial State|A|A|false|
|New assignment applied (callback running)|A + B|A|true|
|After assignment callback failure handling|A + B|A|false|
|After the mismatch exception on the next attempt|A + B|A|true|
|Subsequent attempts|A + B|A|Remains true
(reconciliation returns immediately)|
The same issue can occur when an assignment change includes revocation of
existing tasks. For example, if A is revoked and the consumer assignment is
changed to B before B’s assignment callback fails, the consumer assignment
remains B while the completed assignment remains A. The next attempt encounters
the same mismatch.
This is particularly problematic for exceptions such as
{{{}TaskCorruptedException{}}}, which can be handled by an upper layer while
execution continues. The membership manager can remain stuck independently of
recovery from the original error.
*Impact*
The central issue is that an initial callback failure can escalate into a
persistent state error that blocks all subsequent assignment processing.
* *Further polling does not restore progress.* The retained progress flag
prevents execution from reaching the reconciliation logic.
* *New assignments cannot be processed.* Even if the broker provides a
different target assignment, reconciliation remains blocked by the same flag.
* *Processing of newly assigned partitions may never start while this state
persists.* Partitions awaiting successful callback completion remain blocked
from fetching.
* *A live process does not imply assignment progress.* Thread execution and
heartbeats may continue after error handling while reconciliation remains
stalled.
As a result, even a transient or recoverable task error can prevent the member
from resuming processing through subsequent assignment changes.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
