@ParameterizedTest
    @ValueSource(strings = {"zk", "kraft"})
    public void testConfigPreservationAcrossPartitionAlteration(String quorum) throws Exception {
        String cleanUpPolicy = "compact";
        Properties topicConfig = new Properties();
        topicConfig.put(TopicConfig.CLEANUP_POLICY_CONFIG, cleanUpPolicy);
        TestUtils.createTopicWithAdmin(adminClient, testTopicName, scalaBrokers, scalaControllers, defaultNumPartitions, defaultReplicationFactor,
                scala.collection.immutable.Map$.MODULE$.empty(), topicConfig
        );

        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, testTopicName);
        Config props = adminClient.describeConfigs(Collections.singleton(configResource)).all().get().get(configResource);
        // val props = adminZkClient.fetchEntityConfig(ConfigType.Topic, testTopicName)
        assertNotNull(props.get(TopicConfig.CLEANUP_POLICY_CONFIG), "Properties after creation don't contain " + cleanUpPolicy);
        assertEquals(cleanUpPolicy, props.get(TopicConfig.CLEANUP_POLICY_CONFIG).value(), "Properties after creation have incorrect value");

        // pre-create the topic config changes path to avoid a NoNodeException
        if (!isKRaftTest()) {
            zkClient().makeSurePersistentPathExists(kafka.zk.ConfigEntityChangeNotificationZNode.path());
        }

        // modify the topic to add new partitions
        int numPartitionsModified = 3;
        TopicCommand.TopicCommandOptions alterOpts = buildTopicCommandOptionsWithBootstrap("--alter",
            "--partitions", Integer.toString(numPartitionsModified), "--topic", testTopicName);
        topicService.alterTopic(alterOpts);

        TestUtils.waitForAllReassignmentsToComplete(adminClient, 100L);
        Config newProps = adminClient.describeConfigs(Collections.singleton(configResource)).all().get().get(configResource);
        assertNotNull(newProps.get(TopicConfig.CLEANUP_POLICY_CONFIG), "Updated properties do not contain " + TopicConfig.CLEANUP_POLICY_CONFIG);
        assertEquals(cleanUpPolicy, newProps.get(TopicConfig.CLEANUP_POLICY_CONFIG).value(), "Updated properties have incorrect value");
    }
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