private static MetadataResponse rackAwareMetadata(int numNodes,
                                                      List<String> racks,
                                                      Map<String, List<List<Integer>>> partitionReplicas) {
        final List<Node> nodes = new ArrayList<>(numNodes);
        for (int i = 0; i < numNodes; i++)
            nodes.add(new Node(i, "localhost", 1969 + i, racks.get(i % racks.size())));

        List<MetadataResponse.TopicMetadata> topicMetadata = new ArrayList<>();
        for (Map.Entry<String, List<List<Integer>>> topicPartitionCountEntry : partitionReplicas.entrySet()) {
            String topic = topicPartitionCountEntry.getKey();
            int numPartitions = topicPartitionCountEntry.getValue().size();

            List<MetadataResponse.PartitionMetadata> partitionMetadata = new ArrayList<>(numPartitions);
            for (int i = 0; i < numPartitions; i++) {
                TopicPartition tp = new TopicPartition(topic, i);
                List<Integer> replicaIds = topicPartitionCountEntry.getValue().get(i);
                partitionMetadata.add(new PartitionMetadata(
                        Errors.NONE, tp, Optional.of(replicaIds.get(0)), Optional.empty(),
                        replicaIds, replicaIds, Collections.emptyList()));
            }

            topicMetadata.add(new MetadataResponse.TopicMetadata(Errors.NONE, topic, Uuid.ZERO_UUID,
                    Topic.isInternal(topic), partitionMetadata, MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED));
        }

        return RequestTestUtils.metadataResponse(nodes, "kafka-cluster", 0, topicMetadata, ApiKeys.METADATA.latestVersion());
    }
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