KAFKA-16933: New consumer unsubscribe close commit fixes #16272

lianetm · 2024-06-10T18:33:15Z

Fixes for the leave group flow (unsubscribe/close):

fix to send Heartbeat to leave group on close even if the callbacks fail
fix to ensure that if a member gets fenced while blocked on callbacks (ex. on unsubscribe), it will clear its epoch to not include it in commit requests
fix to avoid race on the subscription state object on unsubscribe, updating it only on the background thread when the callbacks to leave complete (success or failure).

Also improving logging in this area.

This reverts commit bfae065.

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

lucasbru

Hey @lianetm. I left a few comments

lucasbru · 2024-06-13T10:47:41Z

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

-                applicationEventHandler.addAndGet(new LeaveOnCloseEvent(calculateDeadlineMs(timer)));
-            },
-            "Failed to send leaveGroup heartbeat with a timeout(ms)=" + timer.timeoutMs(), firstException);
+        completeQuietly(() -> maybeRevokePartitions(),


So the point here seems to be, if maybeRevokePartitions fails, likely due to rebalancelistener, still leave the group.

It looks like a good change to me. I'm just surprised to see that the legacy consumer does not seem to do this? If onPrepareLeave in AbstractCoordinator fails, we won't reach maybeLeaveGroup.

So is this a bug also in the legacy consumer?

You got it right, in this case the legacy consumer won't send a leave group either. We discussed it with @dajac and it seemed right to ensure the new consumer leaves on close (even if the callbacks fail). No changes on the legacy for now though (would require more thought).

lucasbru · 2024-06-13T10:53:24Z

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

@@ -1324,6 +1322,7 @@ void completeQuietly(final Utils.ThrowingRunnable function,
        } catch (TimeoutException e) {
            log.debug("Timeout expired before the {} operation could complete.", msg);


Should we update firstException here?

Totally, good catch.

lucasbru · 2024-06-13T11:11:31Z

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

+                consumer.close(Duration.ZERO);
+            } catch (Exception e) {
+                // best effort to clean up after each test, but may throw (ex. if callbacks where
+                // throwing errors)


I did this before but was asked to not do it, see https://github.com/apache/kafka/pull/15613/files/9fb917e4b1e60f238183c92d1ad3bc2565a7e1ea#r1559907295. That's why I added a "clean-up close" in the tests where close fails, with an expected exception (search for "clean-up" in this file).

I'd also be fine with your (and my original approach) to have a best-effort clean up and ignore exceptions here. But then, let's remove the "clean-up close" code in the other tests. Any consistent approach is fine with me here.

Oh I see, good point. Agree with going with a consistent approach here, but could I bring it in a follow-up PR right after this? (it wouldn't need to get into 3.8). The trick is that now the close may throw for any of the tests that throw on the callback, so I would prefer to play safe, make sure the close does not throw here, and I'll review all the tests that we expect could rightfully throw on close now (Also the other clean-up logic has some Timeout related checks so better not removing blindly, could have some value it seems)

lucasbru · 2024-06-13T11:14:24Z

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

            // Clear the subscription, no matter if the callback execution failed or succeeded.
+            subscriptions.unsubscribe();
            clearSubscription();


Is the name clearSubscription misleading? It seems like it clears the assignment, not the subscription.

you're right, it's clearing assignments, renamed.

lucasbru

Sorry, didn't mean to approve yet

lianetm · 2024-06-13T12:56:14Z

Hey @lucasbru, all comments addressed. Thanks for the review!

lianetm · 2024-06-13T16:46:18Z

Hey @lucasbru , note I included this small commit 56ac969 we realized we needed to avoid the race on unsubscribe/close. I will follow up on another PR with the fix to avoid getting to a revocation of partitions that are not assigned on close, but still having the sanity check at the low level seems a safe bet while the root cause is fixed. cc. @dajac

lucasbru

LGTM. I'll wait for the tests to pass

lianetm · 2024-06-14T00:57:25Z

Build completed with 5 unrelates failures:

Build / JDK 21 and Scala 2.13 / testFenceMultipleBrokers() – org.apache.kafka.controller.QuorumControllerTest
Build / JDK 21 and Scala 2.13 / shouldQuerySpecificStalePartitionStores() – org.apache.kafka.streams.integration.StoreQueryIntegrationTestBuild / JDK 8 and Scala 2.12 / shouldRestoreState() – org.apache.kafka.streams.integration.PurgeRepartitionTopicIntegrationTest
Build / JDK 8 and Scala 2.12 / shouldRestoreState() – org.apache.kafka.streams.integration.PurgeRepartitionTopicIntegrationTest
Build / JDK 8 and Scala 2.12 / testDescribeQuorumStatusSuccessful [6] Type=Raft-Combined, MetadataVersion=4.0-IV0,Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest

philipnee

hi @lianetm thanks for the pr and fixing the logs and issues. I guess we will file a separated issue to address the interrupted exception weve see during testing.

Fixes for the leave group flow (unsubscribe/close): Fix to send Heartbeat to leave group on close even if the callbacks fail fix to ensure that if a member gets fenced while blocked on callbacks (ex. on unsubscribe), it will clear its epoch to not include it in commit requests fix to avoid race on the subscription state object on unsubscribe, updating it only on the background thread when the callbacks to leave complete (success or failure). Also improving logging in this area. Reviewers: Lucas Brutschy <lbrutschy@confluent.io>, Philip Nee <pnee@confluent.io>

lucasbru · 2024-06-14T11:12:36Z

Test failures are unrelated

Merged to trunk and cherrypicked to 3.8 cc @jlprat

lianetm · 2024-06-14T11:44:01Z

Hey @philipnee, agreed there is still an issue with the Interrupt discovered with the stress tests, it's just that as discussed offline that will come in a separate PR (just to not keep stretching this one). Just for the record, there is also another issue similar to the one fixed on this PR for the subscription state https://issues.apache.org/jira/browse/KAFKA-16954. I plan to open a PR today with both fixes for this related blocker. cc. @lucasbru , @jlprat, @dajac . Thanks!

lianetm added 5 commits June 10, 2024 14:28

Changes & improved logs

27fdd0d

Update test to send leave on close if callbacks fail

a1f17a6

Fix log clause

3fadad7

Clear epoch to include on commit when member leaves

bfae065

More logs for commit epoch

a44e2ac

lianetm changed the title ~~[WIP]: New consumer unsubscribe close fixes~~ [WIP]: New consumer unsubscribe close commit fixes Jun 10, 2024

lianetm added 7 commits June 10, 2024 16:05

Revert "Clear epoch to include on commit when member leaves"

19b7e9d

This reverts commit bfae065.

More logs

51d1bf9

Fix style

e46e071

Fix for fencing while running callbacks

238d907

Merge branch 'trunk' into consumer-unsubscribe-close

20fc8a8

fix test

64327dd

Update subscription state in background on unsubscribe

fe5ddf6

lianetm changed the title ~~[WIP]: New consumer unsubscribe close commit fixes~~ KAFKA-16933: New consumer unsubscribe close commit fixes Jun 11, 2024

lianetm marked this pull request as ready for review June 11, 2024 18:29

Merge branch 'trunk' into consumer-unsubscribe-close

3dcd2b1

lianetm commented Jun 12, 2024

View reviewed changes

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java Show resolved Hide resolved

Remove logs

46c75fc

lucasbru self-requested a review June 13, 2024 08:29

lucasbru approved these changes Jun 13, 2024

View reviewed changes

lucasbru requested changes Jun 13, 2024

View reviewed changes

Fixes addressing comments

5a4402b

lianetm requested a review from lucasbru June 13, 2024 12:58

Validate revocation of assigned only

56ac969

lucasbru approved these changes Jun 13, 2024

View reviewed changes

philipnee reviewed Jun 14, 2024

View reviewed changes

lucasbru merged commit 46714db into apache:trunk Jun 14, 2024
1 check failed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-16933: New consumer unsubscribe close commit fixes #16272

KAFKA-16933: New consumer unsubscribe close commit fixes #16272

lianetm commented Jun 10, 2024 •

edited

Loading

lucasbru left a comment

lucasbru Jun 13, 2024

lianetm Jun 13, 2024

lucasbru Jun 13, 2024

lianetm Jun 13, 2024

lucasbru Jun 13, 2024

lianetm Jun 13, 2024 •

edited

Loading

lucasbru Jun 13, 2024

lianetm Jun 13, 2024

lucasbru left a comment

lianetm commented Jun 13, 2024

lianetm commented Jun 13, 2024

lucasbru left a comment

lianetm commented Jun 14, 2024

philipnee left a comment •

edited

Loading

lucasbru commented Jun 14, 2024

lianetm commented Jun 14, 2024 •

edited

Loading

		@@ -1324,6 +1322,7 @@ void completeQuietly(final Utils.ThrowingRunnable function,
		} catch (TimeoutException e) {
		log.debug("Timeout expired before the {} operation could complete.", msg);

KAFKA-16933: New consumer unsubscribe close commit fixes #16272

KAFKA-16933: New consumer unsubscribe close commit fixes #16272

Conversation

lianetm commented Jun 10, 2024 • edited Loading

lucasbru left a comment

Choose a reason for hiding this comment

lucasbru Jun 13, 2024

Choose a reason for hiding this comment

lianetm Jun 13, 2024

Choose a reason for hiding this comment

lucasbru Jun 13, 2024

Choose a reason for hiding this comment

lianetm Jun 13, 2024

Choose a reason for hiding this comment

lucasbru Jun 13, 2024

Choose a reason for hiding this comment

lianetm Jun 13, 2024 • edited Loading

Choose a reason for hiding this comment

lucasbru Jun 13, 2024

Choose a reason for hiding this comment

lianetm Jun 13, 2024

Choose a reason for hiding this comment

lucasbru left a comment

Choose a reason for hiding this comment

lianetm commented Jun 13, 2024

lianetm commented Jun 13, 2024

lucasbru left a comment

Choose a reason for hiding this comment

lianetm commented Jun 14, 2024

philipnee left a comment • edited Loading

Choose a reason for hiding this comment

lucasbru commented Jun 14, 2024

lianetm commented Jun 14, 2024 • edited Loading

lianetm commented Jun 10, 2024 •

edited

Loading

lianetm Jun 13, 2024 •

edited

Loading

philipnee left a comment •

edited

Loading

lianetm commented Jun 14, 2024 •

edited

Loading