KAFKA-19335: Membership managers send negative epoch in JOINING #19818

lucasbru · 2025-05-26T16:08:33Z

There is a sequence of interactions with the membership managers of
KIP-848, KIP-932, KIP-1071 that can put the membership manager into
JOINING state, but where member epoch is set to -1. This can result in
an invalid request being sent, since joining heartbeats should not have
member epoch -1. This may lead to the member failing to join. In the
case of streams, the group coordinator will return INVALID_REQUEST.

This is the sequence triggering the bug, which seems to relatively
likely, caused by two heartbeat responses being received after the next
one has been sent.

membershipManager.leaveGroup();
    -> transitions to LEAVING
membershipManager.onHeartbeatRequestGenerated();
    -> transitions to UNSUBSCRIBED
membershipManager.onHeartbeatSuccess(... with member epoch > 0);
    -> unblocks the consumer
membershipManager.onSubscriptionUpdated();
membershipManager.onConsumerPoll();
    -> transitions to JOINING
membershipManager.onHeartbeatSuccess(... with member epoch < 0);
    -> updates the epoch to a negative value

Now we are in state JOINING with memberEpoch=-1, and the next
heartbeat we send will be malformed, triggering INVALID_REQUEST

The bug may also be triggered if the unsubscribe times out, but this
seems more of a corner case.

To prevent the bug, we are taking two measures: The likely path to
triggering the bug can be prevented by not unblocking an unsubscribe
call in the consumer when a non-leave-heartbeat epoch is received. Once
we have sent out leave group heartbeat, we will ignore all heartbeats,
except for those containing memberEpoch < 0.

For extra measure, we also prevent the second case (unsubscribe timing
out). In this case, the consumer gets unblocked before we have received
the leave group heartbeat response, and may resubscribe to the group. In
this case, we shall just ignore the heartbeat response that contains a
member epoch < 0, once it arrives, and we have already left the
UNSUBSCRIBED state.

Reviewers: Lianet Magrans [email protected], Andrew Schofield
[email protected], Shivsundar R [email protected]

lianetm · 2025-05-26T18:09:26Z

Hey @lucasbru, thanks for taking on this one. Agree with the gap on leave HB responses received in unexpected order. And the fix makes sense to me (only complete the leave if the HB response is a response to leave, and never apply epoch received in a leave HB response).

What I'm not seeing clearly is how this would lead to INVALID_REQUEST? (so worried that even though this is a sensible gap and fix there may still be something else behind the failure you got?). If this race happens, I expect that we end up sending a full HB (all fields), but with the -1 epoch, correct? Then the request should fail with UNKNOWN_MEMBER, the moment the coordinator tries to find the member that wants to leave

kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

Line 3902 in 6e380fb

StreamsGroupMember member = group.getMemberOrThrow(memberId);

(same applies for the Consumer btw, we should get UNKNOWN_MEMBER if the client sends a full HB to join but with epoch -1 by mistake/race).

Thoughts? not sure if I'm missing something here

lucasbru · 2025-05-26T18:17:37Z

Thoughts? not sure if I'm missing something here

Thanks for taking a look, @lianetm ! You are right, consumer groups and share groups should fail with UNKNOWN_MEMBER_ID. In streams, we send the topology in a full request, and have code to reject when it is being sent in anything but a joining heartbeat. This will trigger the INVALID_REQUEST. Hope that will clear things up!

https://github.com/apache/kafka/blob/trunk/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java#L539

So indeed this will mean that we get fenced in share/consumer, instead of running into a fatal error as in streams. So it's less of a concern (but I suppose still good to fix) for non-streams groups.

ShivsundarR · 2025-05-27T06:51:04Z

...ts/src/test/java/org/apache/kafka/clients/consumer/internals/ShareMembershipManagerTest.java

+        membershipManager.onHeartbeatRequestGenerated();
+        assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+
+        // Receive a previous heartbeat response, which should be ignored


This response is not "ignored" right, we are setting the member epoch in the response to membershipManager.memberEpoch (-1 in this case). The leave operation is then processed successfully post which we are asserting if leaveResult has completed.
If this is the case, can we change the comment above to reflect the same. (and in the other tests too).
Thanks.

Good point. The comment didn't match what was being tested. I updated the test to include both a response that is ignored, and a response that actually unblocks the consumer.

ShivsundarR

Thanks @lucasbru for the PR. Just one minor question in the tests, changes look good overall.

AndrewJSchofield

Thanks for the PR. Looks good to me from the point of view of share groups.

AndrewJSchofield · 2025-05-27T09:48:03Z

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

            log.debug("Member {} with epoch {} received a successful response to the heartbeat " +
-                    "to leave the group and completed the leave operation. ", memberId, memberEpoch);
+                "to leave the group and completed the leave operation. ", memberId, memberEpoch);


nit: The indentation of this line is now inconsistent with the rest of the method.

lucasbru · 2025-05-27T17:04:58Z

@ShivsundarR @lianetm ready for re-review

ShivsundarR

LGTM ! Thanks @lucasbru.

lianetm

Just one question on tests, thanks!

lianetm · 2025-05-28T11:12:39Z

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

+        membershipManager.onHeartbeatSuccess(new ConsumerGroupHeartbeatResponse(
+            new ConsumerGroupHeartbeatResponseData()
+                .setErrorCode(Errors.NONE.code())
+                .setMemberId(membershipManager.memberId())
+                .setMemberEpoch(MEMBER_EPOCH)
+        ));
+        assertFalse(leaveResult.isDone());


the response being ignored is a non-leave response with MEMBER_EPOCH=1, but the test name "IgnoreLeaveResponse" doesn't align right? Clarifying because we could also want to test that a leave response is ignored (if the member already transitioned to JOINING when the leave resp is received)

In this line, I am leaving the group. The IgnoreLeaveResponse referes to the fact that I ignore the heartbeat after leaving the group is further down (the second membershipManager.onHeartbeatSuccess(createConsumerGroupLeaveResponse(membershipManager.memberId()));).

The IgnoreLeaveResponse referes to the fact that I ignore the heartbeat after leaving the group

Ok ok, not leave response (epoch -1), but response after leave, all good. Thanks!

lianetm

Thanks! LGTM

implementation

5bd9396

github-actions bot added consumer clients labels May 26, 2025

lucasbru changed the title ~~KAFKA-19335: Membership Managers end negative epoch in JOINING~~ KAFKA-19335: Membership managers send negative epoch in JOINING May 26, 2025

lucasbru added KIP-1071 PRs related to KIP-1071 KIP-848 The Next Generation of the Consumer Rebalance Protocol KIP-932 Queues for Kafka labels May 26, 2025

lucasbru requested review from cadonna, lianetm and apoorvmittal10 May 26, 2025 16:19

apoorvmittal10 requested review from AndrewJSchofield and ShivsundarR May 26, 2025 22:09

ShivsundarR reviewed May 27, 2025

View reviewed changes

AndrewJSchofield approved these changes May 27, 2025

View reviewed changes

address comments

17355e5

ShivsundarR approved these changes May 28, 2025

View reviewed changes

lianetm reviewed May 28, 2025

View reviewed changes

lianetm approved these changes May 28, 2025

View reviewed changes

lucasbru merged commit 0c116c9 into apache:trunk May 28, 2025
30 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

KAFKA-19335: Membership managers send negative epoch in JOINING #19818

KAFKA-19335: Membership managers send negative epoch in JOINING #19818

lucasbru commented May 26, 2025 •

edited by github-actions bot

Loading

Uh oh!

lianetm commented May 26, 2025

Uh oh!

lucasbru commented May 26, 2025 •

edited

Loading

Uh oh!

ShivsundarR May 27, 2025 •

edited

Loading

Uh oh!

lucasbru May 27, 2025

Uh oh!

ShivsundarR left a comment

Uh oh!

AndrewJSchofield left a comment

Uh oh!

AndrewJSchofield May 27, 2025

Uh oh!

lucasbru May 27, 2025

Uh oh!

lucasbru commented May 27, 2025

Uh oh!

ShivsundarR left a comment

Uh oh!

lianetm left a comment

Uh oh!

lianetm May 28, 2025

Uh oh!

lucasbru May 28, 2025

Uh oh!

lianetm May 28, 2025 •

edited

Loading

Uh oh!

lianetm left a comment

Uh oh!

Uh oh!

Uh oh!

KAFKA-19335: Membership managers send negative epoch in JOINING #19818

KAFKA-19335: Membership managers send negative epoch in JOINING #19818

Conversation

lucasbru commented May 26, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

lianetm commented May 26, 2025

Uh oh!

lucasbru commented May 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ShivsundarR May 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lucasbru May 27, 2025

Choose a reason for hiding this comment

Uh oh!

ShivsundarR left a comment

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield left a comment

Choose a reason for hiding this comment

Uh oh!

AndrewJSchofield May 27, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru May 27, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru commented May 27, 2025

Uh oh!

ShivsundarR left a comment

Choose a reason for hiding this comment

Uh oh!

lianetm left a comment

Choose a reason for hiding this comment

Uh oh!

lianetm May 28, 2025

Choose a reason for hiding this comment

Uh oh!

lucasbru May 28, 2025

Choose a reason for hiding this comment

Uh oh!

lianetm May 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

lucasbru commented May 26, 2025 •

edited by github-actions bot

Loading

lucasbru commented May 26, 2025 •

edited

Loading

ShivsundarR May 27, 2025 •

edited

Loading

lianetm May 28, 2025 •

edited

Loading