@Test
    public void testPrepareJoinAndRejoinAfterFailedRebalance() {
        final List<TopicPartition> partitions = singletonList(t1p);
        try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, false, Optional.of("group-id"), true)) {
            coordinator.ensureActiveGroup();

            prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.REBALANCE_IN_PROGRESS);

            assertThrows(RebalanceInProgressException.class, () -> coordinator.commitOffsetsSync(
                singletonMap(t1p, new OffsetAndMetadata(100L)),
                time.timer(Long.MAX_VALUE)));

            assertFalse(client.hasPendingResponses());
            assertFalse(client.hasInFlightRequests());

            int generationId = 42;
            String memberId = "consumer-42";

            client.prepareResponse(joinGroupFollowerResponse(generationId, memberId, "leader", Errors.NONE));

            MockTime time = new MockTime(1);

            // onJoinPrepare will be executed and onJoinComplete will not.
            boolean res = coordinator.joinGroupIfNeeded(time.timer(100));

            assertFalse(res);
            assertFalse(client.hasPendingResponses());
            // SynGroupRequest not responded.
            assertEquals(1, client.inFlightRequestCount());
            assertEquals(generationId, coordinator.generation().generationId);
            assertEquals(memberId, coordinator.generation().memberId);

            // Imitating heartbeat thread that clears generation data.
            coordinator.maybeLeaveGroup("Clear generation data.");

            assertEquals(AbstractCoordinator.Generation.NO_GENERATION, coordinator.generation());

            client.respond(syncGroupResponse(partitions, Errors.NONE));

            // Join future should succeed but generation already cleared so result of join is false.
            res = coordinator.joinGroupIfNeeded(time.timer(1));

            assertFalse(res);

            // should have retried sending a join group request already
            assertFalse(client.hasPendingResponses());
            assertEquals(1, client.inFlightRequestCount());

            // Retry join should then succeed
            client.respond(joinGroupFollowerResponse(generationId, memberId, "leader", Errors.NONE));
            client.prepareResponse(syncGroupResponse(partitions, Errors.NONE));

            res = coordinator.joinGroupIfNeeded(time.timer(3000));

            assertTrue(res);
            assertFalse(client.hasPendingResponses());
            assertFalse(client.hasInFlightRequests());
        }
        Collection<TopicPartition> lost = getLost(partitions);
        assertEquals(lost.isEmpty() ? null : lost, rebalanceListener.lost);
        assertEquals(lost.size(), rebalanceListener.lostCount);
    }
Verifies that subscription change updates SubscriptionState correctly even after JoinGroup failures
that don't re-invoke onJoinPrepare.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java