@Test
    public void testCommitOffsetIllegalGenerationShouldResetGenerationId() {
        subscriptions.subscribe(singleton(topic1), Optional.of(rebalanceListener));
        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
        coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));

        client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE));
        client.prepareResponse(syncGroupResponse(Collections.emptyList(), Errors.NONE));

        coordinator.joinGroupIfNeeded(time.timer(Long.MAX_VALUE));

        prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.ILLEGAL_GENERATION);
        RequestFuture<Void> future = coordinator.sendOffsetCommitRequest(singletonMap(t1p,
            new OffsetAndMetadata(100L, "metadata")));

        assertTrue(consumerClient.poll(future, time.timer(30000)));

        assertEquals(AbstractCoordinator.Generation.NO_GENERATION.generationId, coordinator.generation().generationId);
        assertEquals(AbstractCoordinator.Generation.NO_GENERATION.protocolName, coordinator.generation().protocolName);
        // member ID should not be reset
        assertEquals(consumerId, coordinator.generation().memberId);
    }
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