private void supportStableFlag(final short upperVersion, final boolean expectThrows) {
        ConsumerCoordinator coordinator = new ConsumerCoordinator(
            rebalanceConfig,
            new LogContext(),
            consumerClient,
            assignors,
            metadata,
            subscriptions,
            new Metrics(time),
            consumerId + groupId,
            time,
            false,
            autoCommitIntervalMs,
            null,
            true,
            null,
            Optional.empty());

        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
        client.setNodeApiVersions(NodeApiVersions.create(ApiKeys.OFFSET_FETCH.id, (short) 0, upperVersion));

        long offset = 500L;
        String metadata = "blahblah";
        Optional<Integer> leaderEpoch = Optional.of(15);
        OffsetFetchResponse.PartitionData data = new OffsetFetchResponse.PartitionData(offset, leaderEpoch,
            metadata, Errors.NONE);

        if (upperVersion < 8) {
            client.prepareResponse(new OffsetFetchResponse(Errors.NONE, singletonMap(t1p, data)));
        } else {
            client.prepareResponse(offsetFetchResponse(Errors.NONE, singletonMap(t1p, data)));
        }
        if (expectThrows) {
            assertThrows(UnsupportedVersionException.class,
                () -> coordinator.fetchCommittedOffsets(singleton(t1p), time.timer(Long.MAX_VALUE)));
        } else {
            Map<TopicPartition, OffsetAndMetadata> fetchedOffsets = coordinator.fetchCommittedOffsets(singleton(t1p),
                time.timer(Long.MAX_VALUE));

            assertNotNull(fetchedOffsets);
            assertEquals(new OffsetAndMetadata(offset, leaderEpoch, metadata), fetchedOffsets.get(t1p));
        }
    }
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