Repository: apache/kafka
File: tools/src/test/java/org/apache/kafka/tools/ConfigCommandIntegrationTest.java (Java)

Code before:
            );
        }
    }

    // Test case from KAFKA-13788
    @ClusterTest(serverProperties = {
        // Must be at greater than 1MB per cleaner thread, set to 2M+2 so that we can set 2 cleaner threads.
        @ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "2097154"),
    })
    public void testUpdateBrokerConfigNotAffectedByInvalidConfig() {
        try (Admin client = cluster.admin()) {
            ConfigCommand.alterConfig(client, new ConfigCommand.ConfigCommandOptions(
                    toArray(asList("--bootstrap-server", cluster.bootstrapServers(),
                            "--alter",
                            "--add-config", "log.cleaner.threadzz=2",
                            "--entity-type", "brokers",
                            "--entity-default"))));

            ConfigCommand.alterConfig(client, new ConfigCommand.ConfigCommandOptions(
                    toArray(asList("--bootstrap-server", cluster.bootstrapServers(),
                            "--alter",
                            "--add-config", "log.cleaner.threads=2",
                            "--entity-type", "brokers",
                            "--entity-default"))));
            kafka.utils.TestUtils.waitUntilTrue(
                    () -> cluster.brokers().values().stream()
                        .map(KafkaBroker::socketServer)
                        .allMatch(broker -> broker.config().getInt("log.cleaner.threads") == 2),
                    () -> "Timeout waiting for topic config propagating to broker",
                    org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS,
                    100L);
        }
    }

    @ClusterTest(
         // Must be at greater than 1MB per cleaner thread, set to 2M+2 so that we can set 2 cleaner threads.
         serverProperties = {@ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "2097154")},
         metadataVersion = MetadataVersion.IBP_3_9_IV0
    )
    public void testUnsupportedVersionException() {
        try (Admin client = cluster.admin()) {
            Admin spyAdmin = Mockito.spy(client);

            AlterConfigsResult mockResult = AdminClientTestUtils.alterConfigsResult(
                    new ConfigResource(ConfigResource.Type.BROKER, ""), new UnsupportedVersionException("simulated error"));
            Mockito.doReturn(mockResult).when(spyAdmin)
                    .incrementalAlterConfigs(anyMap(), any(AlterConfigsOptions.class));
            assertEquals(
                    "The INCREMENTAL_ALTER_CONFIGS API is not supported by the cluster. The API is supported starting from version 2.3.0. You may want to use an older version of this tool to interact with your cluster, or upgrade your brokers to version 2.3.0 or newer to avoid this error.",
                    assertThrows(UnsupportedVersionException.class, () -> {
                        ConfigCommand.alterConfig(spyAdmin, new ConfigCommand.ConfigCommandOptions(

Review comment (suggestion):
```java
                    () -> cluster.brokers().values().stream()
                        .map(KafkaBroker::config)
                        .allMatch(config -> config.getInt("log.cleaner.threads") == 2),
```

Code after:
            );
        }
    }

    // Test case from KAFKA-13788
    @ClusterTest(serverProperties = {
        // Must be at greater than 1MB per cleaner thread, set to 2M+2 so that we can set 2 cleaner threads.
        @ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "2097154"),
    })
    public void testUpdateBrokerConfigNotAffectedByInvalidConfig() {
        try (Admin client = cluster.admin()) {
            ConfigCommand.alterConfig(client, new ConfigCommand.ConfigCommandOptions(
                    toArray(asList("--bootstrap-server", cluster.bootstrapServers(),
                            "--alter",
                            "--add-config", "log.cleaner.threadzz=2",
                            "--entity-type", "brokers",
                            "--entity-default"))));

            ConfigCommand.alterConfig(client, new ConfigCommand.ConfigCommandOptions(
                    toArray(asList("--bootstrap-server", cluster.bootstrapServers(),
                            "--alter",
                            "--add-config", "log.cleaner.threads=2",
                            "--entity-type", "brokers",
                            "--entity-default"))));
            kafka.utils.TestUtils.waitUntilTrue(
                    () -> cluster.brokers().values().stream()
                        .map(KafkaBroker::config)
                        .allMatch(config -> config.getInt("log.cleaner.threads") == 2),
                    () -> "Timeout waiting for topic config propagating to broker",
                    org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS,
                    100L);
        }
    }

    @ClusterTest(
         // Must be at greater than 1MB per cleaner thread, set to 2M+2 so that we can set 2 cleaner threads.
         serverProperties = {@ClusterConfigProperty(key = "log.cleaner.dedupe.buffer.size", value = "2097154")},
         metadataVersion = MetadataVersion.IBP_3_9_IV0
    )
    public void testUnsupportedVersionException() {
        try (Admin client = cluster.admin()) {
            Admin spyAdmin = Mockito.spy(client);

            AlterConfigsResult mockResult = AdminClientTestUtils.alterConfigsResult(
                    new ConfigResource(ConfigResource.Type.BROKER, ""), new UnsupportedVersionException("simulated error"));
            Mockito.doReturn(mockResult).when(spyAdmin)
                    .incrementalAlterConfigs(anyMap(), any(AlterConfigsOptions.class));
            assertEquals(
                    "The INCREMENTAL_ALTER_CONFIGS API is not supported by the cluster. The API is supported starting from version 2.3.0. You may want to use an older version of this tool to interact with your cluster, or upgrade your brokers to version 2.3.0 or newer to avoid this error.",
                    assertThrows(UnsupportedVersionException.class, () -> {
                        ConfigCommand.alterConfig(spyAdmin, new ConfigCommand.ConfigCommandOptions(