private void verifyRackAwareConsumerRebalance(List<List<Integer>> partitionReplicas1,
                                                  List<List<Integer>> partitionReplicas2,
                                                  boolean rackAwareConsumer,
                                                  boolean expectRebalance) {
        List<String> racks = Arrays.asList("rack-a", "rack-b", "rack-c");
        MockPartitionAssignor assignor = partitionAssignor;
        String consumerRackId = null;
        if (rackAwareConsumer) {
            consumerRackId = racks.get(0);
            assignor = new RackAwareAssignor(protocol);
            createRackAwareCoordinator(consumerRackId, assignor);
        }

        MetadataResponse metadataResponse1 = rackAwareMetadata(6, racks, Collections.singletonMap(topic1, partitionReplicas1));
        MetadataResponse metadataResponse2 = rackAwareMetadata(6, racks, Collections.singletonMap(topic1, partitionReplicas2));
        verifyRebalanceWithMetadataChange(Optional.ofNullable(consumerRackId), assignor, metadataResponse1, metadataResponse2, expectRebalance);
    }
Verifies that the consumer re-joins after a metadata change. If JoinGroup fails
and metadata reverts to its original value, the consumer should still retry JoinGroup.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java