Repository: apache/kafka
File: clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareMembershipManagerTest.java (Java)

Code before:
        ShareMembershipManager membershipManager = createMembershipManager();
        when(membershipManager.state()).thenReturn(state);
        ShareGroupHeartbeatResponseData responseData = mock(ShareGroupHeartbeatResponseData.class);

        membershipManager.onHeartbeatSuccess(createShareGroupHeartbeatResponse(responseData));

        assertEquals(state, membershipManager.state());
        verify(responseData, never()).memberId();
        // In unsubscribed, we check if we received a leave group response, so we do verify member epoch.
        if (state != MemberState.UNSUBSCRIBED) {
            verify(responseData, never()).memberEpoch();
        }
        verify(responseData, never()).assignment();
    }

    @Test
    public void testIgnoreLeaveResponseWhenNotLeavingGroup() {
        ShareMembershipManager membershipManager = createMemberInStableState();

        CompletableFuture<Void> leaveResult = membershipManager.leaveGroup();

        // Send leave request, transitioning to UNSUBSCRIBED state
        membershipManager.onHeartbeatRequestGenerated();
        assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());

        // Receive a previous heartbeat response, which should be ignored
        membershipManager.onHeartbeatSuccess(new ShareGroupHeartbeatResponse(
            new ShareGroupHeartbeatResponseData()
                .setErrorCode(Errors.NONE.code())
                .setMemberId(membershipManager.memberId())
                .setMemberEpoch(membershipManager.memberEpoch())
        ));
        assertTrue(leaveResult.isDone());

        // Share unblocks and updates subscription
        membershipManager.onSubscriptionUpdated();
        membershipManager.onConsumerPoll();

        membershipManager.onHeartbeatSuccess(createShareGroupLeaveResponse(membershipManager.memberId()));

        assertEquals(MemberState.JOINING, membershipManager.state());
        assertEquals(0, membershipManager.memberEpoch());
    }

    @Test
    public void testLeaveGroupWhenStateIsReconciling() {
        ShareMembershipManager membershipManager = mockJoinAndReceiveAssignment(false);
        assertEquals(MemberState.RECONCILING, membershipManager.state());

        testLeaveGroupReleasesAssignmentAndResetsEpochToSendLeaveGroup(membershipManager);
    }

Review comment (suggestion):
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.

Code after:
        ShareMembershipManager membershipManager = createMembershipManager();
        when(membershipManager.state()).thenReturn(state);
        ShareGroupHeartbeatResponseData responseData = mock(ShareGroupHeartbeatResponseData.class);

        membershipManager.onHeartbeatSuccess(createShareGroupHeartbeatResponse(responseData));

        assertEquals(state, membershipManager.state());
        verify(responseData, never()).memberId();
        // In unsubscribed, we check if we received a leave group response, so we do verify member epoch.
        if (state != MemberState.UNSUBSCRIBED) {
            verify(responseData, never()).memberEpoch();
        }
        verify(responseData, never()).assignment();
    }

    @Test
    public void testIgnoreLeaveResponseWhenNotLeavingGroup() {
        ShareMembershipManager membershipManager = createMemberInStableState();

        CompletableFuture<Void> leaveResult = membershipManager.leaveGroup();

        // Send leave request, transitioning to UNSUBSCRIBED state
        membershipManager.onHeartbeatRequestGenerated();
        assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());

        // Receive a previous heartbeat response, which should be ignored
        membershipManager.onHeartbeatSuccess(new ShareGroupHeartbeatResponse(
            new ShareGroupHeartbeatResponseData()
                .setErrorCode(Errors.NONE.code())
                .setMemberId(membershipManager.memberId())
                .setMemberEpoch(MEMBER_EPOCH)
        ));
        assertFalse(leaveResult.isDone());

        // Receive a leave heartbeat response, which should unblock the consumer
        membershipManager.onHeartbeatSuccess(createShareGroupLeaveResponse(membershipManager.memberId()));
        assertTrue(leaveResult.isDone());

        // Share unblocks and updates subscription
        membershipManager.onSubscriptionUpdated();
        membershipManager.onConsumerPoll();

        membershipManager.onHeartbeatSuccess(createShareGroupLeaveResponse(membershipManager.memberId()));

        assertEquals(MemberState.JOINING, membershipManager.state());
        assertEquals(0, membershipManager.memberEpoch());
    }

    @Test
    public void testLeaveGroupWhenStateIsReconciling() {
        ShareMembershipManager membershipManager = mockJoinAndReceiveAssignment(false);