@Test
    public void testCommitOffsetShouldNotSetInstanceIdIfMemberIdIsUnknown() {
        rebalanceConfig = buildRebalanceConfig(groupInstanceId);
        ConsumerCoordinator coordinator = buildCoordinator(
            rebalanceConfig,
            new Metrics(),
            assignors,
            false,
            subscriptions
        );

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

        client.prepareResponse(body -> {
            OffsetCommitRequestData data = ((OffsetCommitRequest) body).data();
            return data.groupInstanceId() == null && data.memberId().isEmpty();
        }, offsetCommitResponse(Collections.emptyMap()));

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

        assertTrue(consumerClient.poll(future, time.timer(5000)));
        assertFalse(future.failed());
    }
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