@ParameterizedTest
    @ValueSource(strings = {"zk", "kraft"})
    public void testCreateAlterTopicWithRackAware(String quorum) throws Exception {
        Map<Integer, String> rackInfo = new HashMap<>();
        rackInfo.put(0, "rack1");
        rackInfo.put(1, "rack2");
        rackInfo.put(2, "rack2");
        rackInfo.put(3, "rack1");
        rackInfo.put(4, "rack3");
        rackInfo.put(5, "rack3");

        int numPartitions = 18;
        int replicationFactor = 3;
        TestUtils.createTopicWithAdmin(adminClient, testTopicName, scalaBrokers, scalaControllers, numPartitions, replicationFactor,
                scala.collection.immutable.Map$.MODULE$.empty(), new Properties()
        );

        Map<Integer, List<Integer>> assignment = adminClient.describeTopics(Collections.singletonList(testTopicName))
            .allTopicNames().get().get(testTopicName).partitions()
            .stream()
            .collect(Collectors.toMap(
                info -> info.partition(),
                info -> info.replicas().stream().map(Node::id).collect(Collectors.toList())));
        checkReplicaDistribution(assignment, rackInfo, rackInfo.size(), numPartitions,
            replicationFactor, true, true, true);

        int alteredNumPartitions = 36;
        // verify that adding partitions will also be rack aware
        TopicCommand.TopicCommandOptions alterOpts = buildTopicCommandOptionsWithBootstrap("--alter",
            "--partitions", Integer.toString(alteredNumPartitions),
            "--topic", testTopicName);
        topicService.alterTopic(alterOpts);

        TestUtils.waitForAllReassignmentsToComplete(adminClient, 100L);
        kafka.utils.TestUtils.waitUntilTrue(
            () -> brokers().forall(p -> p.metadataCache().getTopicPartitions(testTopicName).size() == alteredNumPartitions),
            () -> "Timeout waiting for new assignment propagating to broker",
            org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS, 100L);

        assignment = adminClient.describeTopics(Collections.singletonList(testTopicName))
            .allTopicNames().get().get(testTopicName).partitions().stream()
            .collect(Collectors.toMap(info -> info.partition(), info -> info.replicas().stream().map(Node::id).collect(Collectors.toList())));
        checkReplicaDistribution(assignment, rackInfo, rackInfo.size(), alteredNumPartitions, replicationFactor,
            true, true, true);
    }
Implementations must override this method to return a set of KafkaConfigs. This method will be invoked for every
test and should not reuse previous configurations unless they select their ports randomly when servers are started.

Note the replica fetch max bytes is set to `1` in order to throttle the rate of replication for test
`testDescribeUnderReplicatedPartitionsWhenReassignmentIsInProgress`.
tools/src/test/java/org/apache/kafka/tools/TopicCommandIntegrationTest.java