@Test
    public void testSameAssignmentReconciledAgainWhenFenced() {
        MembershipManagerImpl membershipManager = createMemberInStableState();
        Uuid topic1 = Uuid.randomUuid();
        final Assignment assignment1 = new ConsumerGroupHeartbeatResponseData.Assignment();
        final Assignment assignment2 = new ConsumerGroupHeartbeatResponseData.Assignment()
            .setTopicPartitions(Collections.singletonList(
                new TopicPartitions()
                    .setTopicId(topic1)
                    .setPartitions(Arrays.asList(0, 1, 2))
            ));
        when(metadata.topicNames()).thenReturn(Collections.singletonMap(topic1, "topic1"));
        assertEquals(toTopicIdPartitionMap(assignment1), membershipManager.currentAssignment().partitions);

        // Receive assignment, wait on commit
        membershipManager.onHeartbeatSuccess(createConsumerGroupHeartbeatResponse(assignment2).data());
        assertEquals(MemberState.RECONCILING, membershipManager.state());
        CompletableFuture<Void> commitResult = new CompletableFuture<>();
        when(commitRequestManager.maybeAutoCommitSyncBeforeRevocation(anyLong())).thenReturn(commitResult);
        membershipManager.poll(time.milliseconds());

        // Get fenced, commit completes
        membershipManager.transitionToFenced();

        assertEquals(MemberState.JOINING, membershipManager.state());
        assertTrue(membershipManager.currentAssignment().isNone());
        assertTrue(subscriptionState.assignedPartitions().isEmpty());

        commitResult.complete(null);

        assertEquals(MemberState.JOINING, membershipManager.state());
        assertTrue(membershipManager.currentAssignment().isNone());
        assertTrue(subscriptionState.assignedPartitions().isEmpty());

        // We have to reconcile & ack the assignment again
        membershipManager.onHeartbeatSuccess(createConsumerGroupHeartbeatResponse(assignment1).data());
        assertEquals(MemberState.RECONCILING, membershipManager.state());
        membershipManager.poll(time.milliseconds());
        assertEquals(MemberState.ACKNOWLEDGING, membershipManager.state());
        membershipManager.onHeartbeatRequestGenerated();
        assertEquals(MemberState.STABLE, membershipManager.state());
        assertEquals(toTopicIdPartitionMap(assignment1), membershipManager.currentAssignment().partitions);
    }
This is the case where a member rejoins (due to fence). If
the member gets the same assignment again, it should still reconcile and ack the assignment.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java