@ParameterizedTest
    @ValueSource(strings = {"zk", "kraft"})
    public void testDescribeUnavailablePartitions(String quorum) throws ExecutionException, InterruptedException {
        int partitions = 6;
        short replicationFactor = 1;
        TestUtils.createTopicWithAdmin(adminClient, testTopicName, scalaBrokers, scalaControllers, partitions, replicationFactor,
                scala.collection.immutable.Map$.MODULE$.empty(), new Properties()
        );
        try {
            // check which partition is on broker 0 which we'll kill
            TopicDescription testTopicDescription = adminClient.describeTopics(Collections.singletonList(testTopicName))
                .allTopicNames().get().get(testTopicName);
            int partitionOnBroker0 = testTopicDescription.partitions().stream()
                .filter(partition -> partition.leader().id() == 0)
                .findFirst().get().partition();

            killBroker(0);

            // wait until the topic metadata for the test topic is propagated to each alive broker
            kafka.utils.TestUtils.waitUntilTrue(
                () -> {
                    boolean result = true;
                    for (KafkaBroker server : JavaConverters.asJavaCollection(brokers())) {
                        if (server.config().brokerId() != 0) {
                            Set<String> topicNames = Collections.singleton(testTopicName);
                            Collection<MetadataResponseData.MetadataResponseTopic> topicMetadatas =
                                JavaConverters.asJavaCollection(server.dataPlaneRequestProcessor().metadataCache()
                                .getTopicMetadata(JavaConverters.asScalaSetConverter(topicNames).asScala().toSet(),
                                    ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
                                    false, false)
                                );
                            Optional<MetadataResponseData.MetadataResponseTopic> testTopicMetadata = topicMetadatas.stream()
                                .filter(metadata -> metadata.name().equals(testTopicName))
                                .findFirst();
                            if (!testTopicMetadata.isPresent()) {
                                throw new AssertionError("Topic metadata is not found in metadata cache");
                            }
                            Optional<MetadataResponseData.MetadataResponsePartition> testPartitionMetadata = testTopicMetadata.get().partitions().stream()
                                .filter(metadata -> metadata.partitionIndex() == partitionOnBroker0)
                                .findFirst();
                            if (!testPartitionMetadata.isPresent()) {
                                throw new AssertionError("Partition metadata is not found in metadata cache");
                            }
                            result = result && testPartitionMetadata.get().errorCode() == Errors.LEADER_NOT_AVAILABLE.code();
                        }
                    }
                    return result;
                },
                () -> String.format("Partition metadata for %s is not propagated", testTopicName),
                org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS, 100L);

            // grab the console output and assert
            String output = captureDescribeTopicStandardOut(buildTopicCommandOptionsWithBootstrap("--describe", "--topic", testTopicName, "--unavailable-partitions"));
            String[] rows = output.split(System.lineSeparator());
            assertTrue(rows[0].startsWith(String.format("Topic: %s", testTopicName)),
                    "Unexpected Topic " + rows[0] + " received. Expect " + String.format("Topic: %s", testTopicName));
            assertTrue(rows[0].contains("Leader: none\tReplicas: 0\tIsr:"),
                    "Rows did not contain 'Leader: none\tReplicas: 0\tIsr:'");
        } finally {
            restartDeadBrokers(false);
        }
    }
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