KAFKA-15592: Allow member to not sending a heartbeat when a group ID is configured #14390

philipnee · 2023-09-14T18:10:41Z

The PR consists of two changes

Replace the GroupState with the membershipManager
Introduce an "inactive" state

We are planning to replace GroupState with MembershipManager as the latter already includes the necessary fields (groupId, groupInstanceId). This patch is a minor refactor of swapping the two components as the follow up of #14364

However, during the patching, we've realized the heartbeat will be sent continuously unless the member enters a fatal state. This needs to be patched up because the member should be able to join/leave a group at anytime during its life cycle.

dajac · 2023-09-19T15:17:38Z

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

@@ -180,19 +180,19 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
    private class OffsetCommitRequestState {
        private final Map<TopicPartition, OffsetAndMetadata> offsets;
        private final String groupId;
-        private final GroupState.Generation generation;
+        private final int memberEpoch;


For my understanding, how do we plan to update the member epoch when stale member epoch is received? Are we going to recreate a new OffsetCommitRequestState?

You are right. Stale member epcoh error isn't retriable, so the commit request would fail. since the request state is tight to individual request, so the subsequent request will get a new state.

Though - When we receive such error - It seems like we should also update membershipManager which is not done here.

Stale member epcoh error isn't retriable

A stale member epoch error is retriable... but it should be retried with the new epoch.

We could do it either way - recreate this RequestState object or resend the same object with the updated epoch.

Update DefaultBackgroundThread.java

…fka into kip-848-group-state-removal

lianetm · 2023-10-11T13:36:38Z

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

-                            .setTopics(new ArrayList<>(requestTopicDataMap.values())));
+            OffsetCommitRequestData data = new OffsetCommitRequestData()
+                    .setGroupId(groupId)
+                    .setGenerationIdOrMemberEpoch(membershipManager.memberEpoch())


We end up taking the epoch, memberId, and groupInstanceId directly from the membershipMgr, even though they are defined as local variables (unused) in this OffsetCommitRequestState class. In the case of the groupId though, we do use the local var. Let's review and make it consistent.

lianetm · 2023-10-11T13:59:35Z

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

+                    config,
+                    coordinatorRequestManager,
+                    membershipManager.get());
+                MembershipManager membershipManager = new MembershipManagerImpl(groupId);


This is redundant here, already created above.

philipnee · 2023-10-12T03:12:06Z

Hello @dajac @lianetm @kirktrue - I think the scope of this PR extended a bit beyond "replacing GroupState" because we've realized the member may join/leave the group anytime during its lifecycle. In particular, when the member hasn't joined/ has already left the group, we should stop sending the heartbeat.

The current implementation assumes the member should always try to join the group if the groupId is provided, due to the fact that the member's initial state is UNJOINED. However, as previously mentioned:

the user may only want to commit offsets
the user may unsubscribe and then assign (so we should keep the consumer alive and allows it to continue to commit offset)
the user might leave the group due to lack of polling (At least I believe that's the current behavior)
The user can unsubscribe, and then subscribe.

In any of the scenarios above, I think we wouldn't want the member to actively send heartbeats to the GC.

As the original intent of the membershipManager is to manage the state only when the member is in a group. However, due to the fact that the member may join/leave the group any time, I feel it might be sensible to always keep an instance active and introduce an inactive state such as NOT_IN_GROUP to indicate the member does not actively try to join a group.

However, it confuses with UNJOINED a bit. I wonder if we should rename the states.

lianetm · 2023-10-13T14:02:13Z

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

        private final CompletableFuture<Void> future;
+        private int memberEpoch;


This should be Optional here too, right? Just as the memberId and groupInstanceId (groupId is the only required one to be able to commit offsets)

lianetm · 2023-10-13T14:03:24Z

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

-                            .setTopics(new ArrayList<>(requestTopicDataMap.values())));
+            OffsetCommitRequestData data = new OffsetCommitRequestData()
+                    .setGroupId(groupId)
+                    .setGenerationIdOrMemberEpoch(memberEpoch)


Aligned with the idea that memberEpoch may not be present, I expect we should only set this if we have one (like it is done for the memberID)

Agreed. Updated.

lianetm · 2023-10-13T14:07:53Z

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

        public OffsetFetchRequestState(final Set<TopicPartition> partitions,
-                                       final GroupState.Generation generation,


We do need member epoch (and member ID), to be included in the request if we have them, exactly as it is done for the OffsetCommit.

My comment is really only to make sure we are on the same page regarding the need of epoch here. But ok if we don't include the changes in this PR, because I also have them in the changes for OffsetFetch v9 that will follow this. I added id and epoch here not only to send it in the request, but also to differentiate the requests that will be retried.

For the current impl, it sends a generation of -1 and id of "". I see both fields are "ignorable", not sure if that means we don't need to send them, @dajac - I assume ignorable means we don't need to send them if we don't have one?

Yes, agree we don't need to send them if we don't have them (member not using the group management capabilities), but the point is member ID/epoch should be here to be included if the member has them.

lianetm · 2023-10-13T14:23:14Z

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

+
+    @Override
+    public void leaveGroup() {
+        resetEpoch();


I don't get the full idea here. What's the intended usage for this? If the intention is to use the transition as a trigger to send the last HB, then resetEpoch is incorrect, because we need epoch -1 (or -2 if static member). If the intention is just to call this after the last HB has been sent, then how are you seeing we'll trigger that last HB? (ex. from the unsubsbcribe)

Maybe I leave it as TODO, because I'm not sure how leaveGroup would integrate as well.

OK, I remember why I wrote this. I think the epoch should never be -1/-2, it is only used for sending a leave group request. When we need/want to leave the group, then I think we would first send the heartbeat with negative epoch, then transition the state to not_in_group, it should also join with epoch = 0.

instead of reset the Epoch, which seems confusing, I made the epcoh = OptionalInt.empty() so that when the member tries to send a commit, it won't send the epoch after leaving the group.

Ensure initial states and states are set correctly after updating make epoch optional

philipnee added the KIP-848 The Next Generation of the Consumer Rebalance Protocol label Sep 14, 2023

dajac reviewed Sep 19, 2023

View reviewed changes

philipnee added 2 commits October 10, 2023 11:59

Remove group state

4dac49e

Update DefaultBackgroundThread.java

5c9b4d2

Update DefaultBackgroundThread.java

philipnee force-pushed the kip-848-group-state-removal branch from e6ab255 to 5c9b4d2 Compare October 10, 2023 18:59

philipnee added 5 commits October 10, 2023 12:05

Remove group state

41c5dbe

Update DefaultBackgroundThread.java

0cdfb89

Update DefaultBackgroundThread.java

clean up and rebase

9beb82c

Merge branch 'kip-848-group-state-removal' of github.com:philipnee/ka…

23a71a4

…fka into kip-848-group-state-removal

Update CommitRequestManager.java

b492ed7

lianetm reviewed Oct 11, 2023

View reviewed changes

philipnee added 4 commits October 11, 2023 09:29

pr comments

950de41

updates

65e6d8b

Update CommitRequestManager.java

1940e99

Fix test by transition the state to UNJOINED

bef4fa0

philipnee changed the title ~~MINOR: Replace GroupState with MembershipManager~~ KAFKA-15592: Allow member to not sending a heartbeat when a group ID is configured Oct 12, 2023

lianetm reviewed Oct 13, 2023

View reviewed changes

Ensure initial states and states are set correctly after updating

bacc17b

Ensure initial states and states are set correctly after updating make epoch optional

lianetm mentioned this pull request Oct 17, 2023

KAFKA-15554: Client state changes for handling one assignment at a time & minor improvements #14413

Merged

philipnee closed this Dec 20, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-15592: Allow member to not sending a heartbeat when a group ID is configured #14390

KAFKA-15592: Allow member to not sending a heartbeat when a group ID is configured #14390

philipnee commented Sep 14, 2023 •

edited

Loading

dajac Sep 19, 2023

philipnee Sep 19, 2023 •

edited

Loading

dajac Oct 10, 2023

philipnee Oct 10, 2023

lianetm Oct 11, 2023

lianetm Oct 11, 2023

philipnee commented Oct 12, 2023

lianetm Oct 13, 2023

lianetm Oct 13, 2023

philipnee Oct 16, 2023

lianetm Oct 13, 2023

lianetm Oct 13, 2023

philipnee Oct 16, 2023

lianetm Oct 17, 2023

lianetm Oct 13, 2023

philipnee Oct 16, 2023

philipnee Oct 16, 2023

philipnee Oct 16, 2023

		private final CompletableFuture<Void> future;
		private int memberEpoch;

		public OffsetFetchRequestState(final Set<TopicPartition> partitions,
		final GroupState.Generation generation,

KAFKA-15592: Allow member to not sending a heartbeat when a group ID is configured #14390

KAFKA-15592: Allow member to not sending a heartbeat when a group ID is configured #14390

Conversation

philipnee commented Sep 14, 2023 • edited Loading

Choose a reason for hiding this comment

philipnee Sep 19, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

philipnee commented Oct 12, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

philipnee commented Sep 14, 2023 •

edited

Loading

philipnee Sep 19, 2023 •

edited

Loading