KAFKA-20428: Fix unsubscribe failure with assignment updates by lianetm · Pull Request #22011 · apache/kafka

lianetm · 2026-04-09T15:46:41Z

Fix to ensure that unsubscribe does not apply any pending assignment
update that may exist in the background queue (e.g, if a reconciliation
completed right before unsubscribing).

Fix by filtering out the assignment update events on unsubscribe (same
approach already done for filtering out error events that the
unsubscribe should not process).

This issue and fix only affects unsubscribe (not close), as unsubscribe
is the only one, other than poll, that processes background events.

Reviewers: Kirk True kirk@kirktrue.pro, Andrew Schofield
aschofield@confluent.io

lianetm · 2026-04-09T15:51:00Z

Just for the record, I filed https://issues.apache.org/jira/browse/KAFKA-20429 as follow-up, to properly review the design so we can simplify and be less error-prone in this area (for now just fixing with the same "filter-out" approach that we already have in place)

lianetm · 2026-04-09T19:05:41Z

@kirktrue / @AndrewJSchofield could you take a look? Thanks!

kirktrue

Thanks for the PR @lianetm!

I left mostly minor comments.

Thanks

kirktrue · 2026-04-09T21:25:24Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

+                // they are not relevant anymore (consumer already unsubscribing).
+                processBackgroundEvents(unsubscribeEvent.future(), timer,
+                    e -> (e instanceof GroupAuthorizationException || e instanceof TopicAuthorizationException),
+                    true);


Since we're already using an anonymous function for Predicate<Exception> ignoreErrorEventException, could we also do that instead of a flag, i.e.:

Suggested change

true);

e -> (event.type() == BackgroundEvent.Type.PARTITIONS_ASSIGNED || event.type() == BackgroundEvent.Type.STREAMS_TASKS_ASSIGNED));

Then we can delete isAssignmentEvent() and the event filtering becomes a little more generalized.

Agree that it would given some consistency in the params, but at the price of complexity in a case where seems to me we don't really need the flexibitliy of a predicate?

This is only used from the unsubscribe to "skip assignment", not different callers filtering different kinds of events. So I would lean towards simple/YAGNI in this case, makes sense?

kirktrue · 2026-04-09T21:30:49Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

+    public void testUnsubscribeWithPendingStreamsTasksAssignedEvent() {
+        consumer = newConsumer(requiredConsumerConfigAndGroupId("consumerGroup"));
+        completeTopicSubscriptionChangeEventSuccessfully();
+        consumer.subscribe(singletonList("topic"));
+        completeUnsubscribeApplicationEventSuccessfully();
+
+        // Add StreamsTasksAssignedEvent to the background queue (simulating an ongoing reconciliation
+        // that completed just before unsubscribe was called)
+        StreamsTasksAssignedEvent assignedEvent = new StreamsTasksAssignedEvent(
+            new TreeSet<>(TOPIC_PARTITION_COMPARATOR), new TreeSet<>(TOPIC_PARTITION_COMPARATOR),
+            new StreamsRebalanceData.Assignment(Set.of(), Set.of(), Set.of(), true));
+        backgroundEventQueue.add(assignedEvent);
+
+        // The call to unsubscribe should complete successfully (StreamsTasksAssignedEvent not processed and completed exceptionally)
+        assertDoesNotThrow(() -> consumer.unsubscribe());
+        verify(applicationEventHandler, never().description("Reconciled assignment updates shouldn't be processed while unsubscribing"))
+                .addAndGet(any(ApplyAssignmentEvent.class));
+        assertTrue(assignedEvent.future().isCompletedExceptionally());
+    }


These unit tests only differ by the type of assignedEvent, right? Could we generalize the test to reduce duplication?

yes, done (consolidated them with params)

kirktrue · 2026-04-09T21:36:03Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

+                        if (event instanceof CompletableEvent) {
+                            ((CompletableEvent<?>) event).future().completeExceptionally(
+                                new KafkaException("Assignment event skipped because consumer is unsubscribing"));
+                        }


Both of these events are CompletableEvents. I guess this is defensive?

yes, just extra check because at this level we don't know about the specific events

lianetm · 2026-04-10T13:20:52Z

Thanks for the review @kirktrue ! Comments addressed

Copilot

Pull request overview

Fixes KAFKA-20428 by preventing AsyncKafkaConsumer#unsubscribe() from processing pending “assignment update” background events that can be queued right before unsubscribe and otherwise cause failures during the unsubscribe flow.

Changes:

Add a skipAssignmentEvents option to AsyncKafkaConsumer#processBackgroundEvents(...) and use it from unsubscribe().
Skip processing PARTITIONS_ASSIGNED / STREAMS_TASKS_ASSIGNED background events during unsubscribe (while still draining the queue).
Update/add unit tests to cover unsubscribe with pending assignment events and adapt existing tests to the new method signature.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
`clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java`	Adds the ability to skip assignment-update background events during unsubscribe and wires it into `unsubscribe()`.
`clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java`	Updates processBackgroundEvents tests for the new signature and adds a parameterized regression test for pending assignment events during unsubscribe.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

AndrewJSchofield

Thanks for the PR. It seems to me that there's a bit of a testing gap because we are completing the assignment events preemptively during unsubscribe and I don't think we have a test to validate. It seems fairly straightforward for the consumer group, but a bit more complicated for the streams group. Do we need to show that the streams group remains in a valid state afterwards?

lianetm · 2026-04-10T21:29:32Z

Makes sense @AndrewJSchofield , added tests to show the recovery at the state machine level (showing members can rejoin after unsubscribe when there was a previous reconciliation clashing with the unsubscribe). Both tests pass locally . Thanks!

lianetm · 2026-04-11T13:42:16Z

@kirktrue I'll go ahead and merge this. If you have any other comments regarding the predicate/boolean param (comment) we can follow-up separately, no need to block the release for that.

Thanks both for the reviews!

Fix to ensure that unsubscribe does not apply any pending assignment update that may exist in the background queue (e.g, if a reconciliation completed right before unsubscribing). Fix by filtering out the assignment update events on unsubscribe (same approach already done for filtering out error events that the unsubscribe should not process). This issue and fix only affects unsubscribe (not close), as unsubscribe is the only one, other than poll, that processes background events. Reviewers: Kirk True <kirk@kirktrue.pro>, Andrew Schofield <aschofield@confluent.io> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

fix & tests

95d1204

github-actions bot added consumer clients labels Apr 9, 2026

kirktrue reviewed Apr 9, 2026

View reviewed changes

consolidate tests

92911e4

github-actions bot added the small Small PRs label Apr 10, 2026

lianetm requested a review from Copilot April 10, 2026 15:32

Copilot started reviewing on behalf of lianetm April 10, 2026 15:33 View session

Copilot AI reviewed Apr 10, 2026

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java Show resolved Hide resolved

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java Outdated Show resolved Hide resolved

update java doc

afbf446

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

github-actions bot removed the small Small PRs label Apr 10, 2026

AndrewJSchofield requested changes Apr 10, 2026

View reviewed changes

more tests

08e2c38

AndrewJSchofield approved these changes Apr 10, 2026

View reviewed changes

lianetm merged commit 4b9eddc into apache:trunk Apr 11, 2026
22 checks passed

	true);
	e -> (event.type() == BackgroundEvent.Type.PARTITIONS_ASSIGNED \|\| event.type() == BackgroundEvent.Type.STREAMS_TASKS_ASSIGNED));

Conversation

lianetm commented Apr 9, 2026 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

lianetm commented Apr 9, 2026

Uh oh!

lianetm commented Apr 9, 2026

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

kirktrue Apr 9, 2026

Choose a reason for hiding this comment

Uh oh!

lianetm Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

kirktrue Apr 9, 2026

Choose a reason for hiding this comment

Uh oh!

lianetm Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

kirktrue Apr 9, 2026

Choose a reason for hiding this comment

Uh oh!

lianetm Apr 10, 2026

Choose a reason for hiding this comment

Uh oh!

lianetm commented Apr 10, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

lianetm commented Apr 10, 2026

Uh oh!

lianetm commented Apr 11, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

lianetm commented Apr 9, 2026 •

edited by github-actions bot

Loading