private ConsumerCoordinator prepareCoordinatorForCloseTest(final boolean useGroupManagement,
                                                               final boolean autoCommit,
                                                               final Optional<String> groupInstanceId,
                                                               final boolean shouldPoll) {
        rebalanceConfig = buildRebalanceConfig(groupInstanceId);
        ConsumerCoordinator coordinator = buildCoordinator(rebalanceConfig,
                                                           new Metrics(),
                                                           assignors,
                                                           autoCommit,
                                                           subscriptions);
        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
        coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));
        if (useGroupManagement) {
            subscriptions.subscribe(singleton(topic1), Optional.of(rebalanceListener));
            client.prepareResponse(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NONE));
            client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE));
            coordinator.joinGroupIfNeeded(time.timer(Long.MAX_VALUE));
        } else {
            subscriptions.assignFromUser(singleton(t1p));
        }

        subscriptions.seek(t1p, 100);
        if (shouldPoll) {
            coordinator.poll(time.timer(Long.MAX_VALUE));
        }

        return coordinator;
    }
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