@ParameterizedTest
    @ValueSource(strings = {"zk", "kraft"})
    public void testDescribeUnderReplicatedPartitionsWhenReassignmentIsInProgress(String quorum) throws ExecutionException, InterruptedException {
        TopicPartition tp = new TopicPartition(testTopicName, 0);

        TestUtils.createTopicWithAdmin(adminClient, testTopicName, scalaBrokers, scalaControllers, defaultNumPartitions, defaultReplicationFactor,
                scala.collection.immutable.Map$.MODULE$.empty(), new Properties()
        );

        // Produce multiple batches.
        TestUtils.generateAndProduceMessages(brokers(), testTopicName, 10, -1);
        TestUtils.generateAndProduceMessages(brokers(), testTopicName, 10, -1);

        // Enable throttling. Note the broker config sets the replica max fetch bytes to `1` upon to minimize replication
        // throughput so the reassignment doesn't complete quickly.
        List<Integer> brokerIds = JavaConverters.seqAsJavaList(brokers()).stream()
            .map(broker -> broker.config().brokerId()).collect(Collectors.toList());

        ToolsTestUtils.setReplicationThrottleForPartitions(adminClient, brokerIds, Collections.singleton(tp), 1);

        TopicDescription testTopicDesc = adminClient.describeTopics(Collections.singleton(testTopicName)).allTopicNames().get().get(testTopicName);
        TopicPartitionInfo firstPartition = testTopicDesc.partitions().get(0);

        List<Integer> replicasOfFirstPartition = firstPartition.replicas().stream().map(Node::id).collect(Collectors.toList());
        List<Integer> replicasDiff = new ArrayList<>(brokerIds);
        replicasDiff.removeAll(replicasOfFirstPartition);
        Integer targetReplica = replicasDiff.get(0);

        adminClient.alterPartitionReassignments(Collections.singletonMap(tp,
            Optional.of(new NewPartitionReassignment(Collections.singletonList(targetReplica))))).all().get();

        // let's wait until the LAIR is propagated
        kafka.utils.TestUtils.waitUntilTrue(
            () -> {
                try {
                    return !adminClient.listPartitionReassignments(Collections.singleton(tp)).reassignments().get()
                        .get(tp).addingReplicas().isEmpty();
                } catch (InterruptedException | ExecutionException e) {
                    throw new RuntimeException(e);
                }
            },
            () -> "Reassignment didn't add the second node",
            org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS, 100L);

        ensureConsistentKRaftMetadata();

        // describe the topic and test if it's under-replicated
        String simpleDescribeOutput = captureDescribeTopicStandardOut(buildTopicCommandOptionsWithBootstrap("--describe", "--topic", testTopicName));
        String[] simpleDescribeOutputRows = simpleDescribeOutput.split(System.lineSeparator());
        assertTrue(simpleDescribeOutputRows[0].startsWith(String.format("Topic: %s", testTopicName)),
                "Unexpected describe output: " + simpleDescribeOutputRows[0]);
        assertEquals(2, simpleDescribeOutputRows.length,
                "Unexpected describe output length: " + simpleDescribeOutputRows.length);

        String underReplicatedOutput = captureDescribeTopicStandardOut(buildTopicCommandOptionsWithBootstrap("--describe", "--under-replicated-partitions"));
        assertEquals("", underReplicatedOutput,
            String.format("--under-replicated-partitions shouldn't return anything: '%s'", underReplicatedOutput));

        int maxRetries = 20;
        long pause = 100L;
        long waitTimeMs = maxRetries * pause;
        AtomicReference<PartitionReassignment> reassignmentsRef = new AtomicReference<>();

        TestUtils.waitUntilTrue(() -> {
            try {
                PartitionReassignment tempReassignments = adminClient.listPartitionReassignments(Collections.singleton(tp)).reassignments().get().get(tp);
                reassignmentsRef.set(tempReassignments);
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException("Error while fetching reassignments", e);
            }
            return reassignmentsRef.get() != null;
        }, () -> "Reassignments did not become non-null within the specified time", waitTimeMs, pause);

        assertFalse(reassignmentsRef.get().addingReplicas().isEmpty());

        ToolsTestUtils.removeReplicationThrottleForPartitions(adminClient, brokerIds, Collections.singleton(tp));
        TestUtils.waitForAllReassignmentsToComplete(adminClient, 100L);
    }
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