@Test
    public void testCommitOffsetRebalanceInProgress() {
        // we cannot retry if a rebalance occurs before the commit completed
        final String consumerId = "leader";

        subscriptions.subscribe(singleton(topic1), Optional.of(rebalanceListener));

        // ensure metadata is up-to-date for leader
        client.updateMetadata(metadataResponse);

        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
        coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));

        // normal join group
        Map<String, List<String>> memberSubscriptions = singletonMap(consumerId, singletonList(topic1));
        partitionAssignor.prepare(singletonMap(consumerId, singletonList(t1p)));

        coordinator.ensureActiveGroup(time.timer(0L));

        assertTrue(coordinator.rejoinNeededOrPending());
        assertNull(coordinator.generationIfStable());

        // when the state is REBALANCING, we would not even send out the request but fail immediately
        assertThrows(RebalanceInProgressException.class, () -> coordinator.commitOffsetsSync(singletonMap(t1p,
            new OffsetAndMetadata(100L, "metadata")), time.timer(Long.MAX_VALUE)));

        final Node coordinatorNode = new Node(Integer.MAX_VALUE - node.id(), node.host(), node.port());
        client.respondFrom(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, Errors.NONE), coordinatorNode);

        client.prepareResponse(body -> {
            SyncGroupRequest sync = (SyncGroupRequest) body;
            return sync.data().memberId().equals(consumerId) &&
                sync.data().generationId() == 1 &&
                sync.groupAssignments().containsKey(consumerId);
        }, syncGroupResponse(singletonList(t1p), Errors.NONE));
        coordinator.poll(time.timer(Long.MAX_VALUE));

        AbstractCoordinator.Generation expectedGeneration = new AbstractCoordinator.Generation(1, consumerId, partitionAssignor.name());
        assertFalse(coordinator.rejoinNeededOrPending());
        assertEquals(expectedGeneration, coordinator.generationIfStable());

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

        assertTrue(coordinator.rejoinNeededOrPending());
        assertEquals(expectedGeneration, coordinator.generationIfStable());
    }
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