@Test
    public void testSubscriptionRackId() {

        String rackId = "rack-a";
        RackAwareAssignor assignor = new RackAwareAssignor(protocol);
        createRackAwareCoordinator(rackId, assignor);

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

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

        Map<String, List<String>> memberSubscriptions = singletonMap(consumerId, singletonList(topic1));
        assignor.prepare(singletonMap(consumerId, singletonList(t1p)));

        client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberSubscriptions, false, Errors.NONE, Optional.of(rackId)));
        client.prepareResponse(syncGroupResponse(singletonList(t1p), Errors.NONE));

        coordinator.poll(time.timer(Long.MAX_VALUE));
        assertEquals(singleton(t1p), coordinator.subscriptionState().assignedPartitions());
        assertEquals(singleton(rackId), assignor.rackIds);
    }
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