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

Code before:
        ConsumerMembershipManager membershipManager = mockJoinAndReceiveAssignment(false);
        assertEquals(MemberState.RECONCILING, membershipManager.state());

        testLeaveGroupReleasesAssignmentAndResetsEpochToSendLeaveGroup(membershipManager);
    }

    @Test
    public void testLeaveGroupWhenMemberOwnsAssignment() {
        Uuid topicId = Uuid.randomUuid();
        String topicName = "topic1";
        ConsumerMembershipManager membershipManager = createMembershipManagerJoiningGroup();
        mockOwnedPartitionAndAssignmentReceived(membershipManager, topicId, topicName, Collections.emptyList());

        receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);

        verifyReconciliationNotTriggered(membershipManager);
        membershipManager.maybeReconcile(true);

        List<TopicIdPartition> assignedPartitions = Arrays.asList(
            new TopicIdPartition(topicId, new TopicPartition(topicName, 0)),
            new TopicIdPartition(topicId, new TopicPartition(topicName, 1)));
        verifyReconciliationTriggeredAndCompleted(membershipManager, assignedPartitions);

        assertEquals(1, membershipManager.currentAssignment().partitions.size());

        testLeaveGroupReleasesAssignmentAndResetsEpochToSendLeaveGroup(membershipManager);
    }

    @Test
    public void testFencedWhenAssignmentEmpty() {
        ConsumerMembershipManager membershipManager = createMemberInStableState();

        // Clear the assignment
        when(subscriptionState.assignedPartitions()).thenReturn(Collections.emptySet());
        when(subscriptionState.hasAutoAssignedPartitions()).thenReturn(false);

        membershipManager.transitionToFenced();

        // Make sure to never call `assignFromSubscribed` again
        verify(subscriptionState, never()).assignFromSubscribed(Collections.emptySet());
    }

    @Test
    public void testLeaveGroupWhenMemberAlreadyLeaving() {
        ConsumerMembershipManager membershipManager = createMemberInStableState();

        // First leave attempt. Should trigger the callbacks and stay LEAVING until
        // callbacks complete and the heartbeat is sent out.
        mockLeaveGroup();
        CompletableFuture<Void> leaveResult1 = membershipManager.leaveGroup();
        verify(subscriptionState).unsubscribe();

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

Code after:
        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() {
        ConsumerMembershipManager 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 ConsumerGroupHeartbeatResponse(
            new ConsumerGroupHeartbeatResponseData()
                .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(createConsumerGroupLeaveResponse(membershipManager.memberId()));

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

        membershipManager.onHeartbeatSuccess(createConsumerGroupLeaveResponse(membershipManager.memberId()));

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

    @Test
    public void testLeaveGroupWhenMemberOwnsAssignment() {
        Uuid topicId = Uuid.randomUuid();
        String topicName = "topic1";
        ConsumerMembershipManager membershipManager = createMembershipManagerJoiningGroup();
        mockOwnedPartitionAndAssignmentReceived(membershipManager, topicId, topicName, Collections.emptyList());

        receiveAssignment(topicId, Arrays.asList(0, 1), membershipManager);

        verifyReconciliationNotTriggered(membershipManager);
        membershipManager.maybeReconcile(true);