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

Code before:
        updateAndCheckInvalidBrokerConfig(Optional.empty());
        updateAndCheckInvalidBrokerConfig(Optional.of(String.valueOf((cluster.brokers().entrySet().iterator().next().getKey()))));
    }

    private void updateAndCheckInvalidBrokerConfig(Optional<String> brokerIdOrDefault) throws InterruptedException {
        List<String> alterOpts = generateDefaultAlterOpts(cluster.bootstrapServers());
        try (Admin client = cluster.admin()) {
            alterConfigWithAdmin(client, brokerIdOrDefault, Map.of("invalid", "2"), alterOpts);
            AtomicReference<String> last = new AtomicReference<>("");

            TestUtils.waitForCondition(() -> {
                Stream<String> describeCommand = Stream.concat(
                        Stream.concat(
                                Stream.of("--bootstrap-server", cluster.bootstrapServers()),
                                Stream.of(entityOp(brokerIdOrDefault).toArray(new String[0]))),
                        Stream.of("--entity-type", "brokers", "--describe")
                );
                String describeResult = captureStandardOut(run(describeCommand));
                last.set(describeResult);

                // We will treat unknown config as sensitive
                // For 'describeResult.contains("sensitive=true")' 
                
                // Sensitive config will not return
                // For 'describeResult.contains("invalid=null")'
                return describeResult.contains("sensitive=true") && describeResult.contains("invalid=null");
            }, 5000, "Dynamic broker config was not visible within 5s. Last describe output:\n" + last.get());
        }
    }

    @ClusterTest
    public void testUpdateInvalidTopicConfigs() throws ExecutionException, InterruptedException {
        List<String> alterOpts = List.of("--bootstrap-server", cluster.bootstrapServers(), "--entity-type", "topics", "--alter");
        try (Admin client = cluster.admin()) {
            client.createTopics(List.of(new NewTopic("test-config-topic", 1, (short) 1))).all().get();
            assertInstanceOf(
                    InvalidConfigurationException.class,
                    assertThrows(
                            ExecutionException.class,
                            () -> ConfigCommand.alterConfig(
                                    client,
                                    new ConfigCommand.ConfigCommandOptions(
                                            toArray(alterOpts,
                                                    List.of("--add-config", "invalid=2", "--entity-type", "topics", "--entity-name", "test-config-topic"))))
                    ).getCause()
            );
        }
    }

    // Test case from KAFKA-13788
    @ClusterTest(serverProperties = {

Review comment (bug):
I guess that it makes sense to wait until we get `invalid=null` in the output, but once that happens we do expect to have `sensitive=true` right away, right? 
If my expectation is right, then we should fix this to ensure we fail fast if sensitive is false for our "invalid" config (without waiting for sensitive to become true)

Code after:
        updateAndCheckInvalidBrokerConfig(Optional.empty());
        updateAndCheckInvalidBrokerConfig(Optional.of(String.valueOf((cluster.brokers().entrySet().iterator().next().getKey()))));
    }

    private void updateAndCheckInvalidBrokerConfig(Optional<String> brokerIdOrDefault) throws InterruptedException {
        List<String> alterOpts = generateDefaultAlterOpts(cluster.bootstrapServers());
        try (Admin client = cluster.admin()) {
            alterConfigWithAdmin(client, brokerIdOrDefault, Map.of("invalid", "2"), alterOpts);
            AtomicReference<String> last = new AtomicReference<>("");

            TestUtils.waitForCondition(() -> {
                Stream<String> describeCommand = Stream.concat(
                        Stream.concat(
                                Stream.of("--bootstrap-server", cluster.bootstrapServers()),
                                Stream.of(entityOp(brokerIdOrDefault).toArray(new String[0]))),
                        Stream.of("--entity-type", "brokers", "--describe")
                );
                String describeResult = captureStandardOut(run(describeCommand));
                last.set(describeResult);

                return describeResult.contains("invalid=null");
            }, 5000, () -> "Dynamic broker config was not visible within 5s (missing 'invalid=null').\n" + 
                    "Last describe output:\n" + last.get());

            assertTrue(last.get().contains("sensitive=true"));
        }
    }

    @ClusterTest
    public void testUpdateInvalidTopicConfigs() throws ExecutionException, InterruptedException {
        List<String> alterOpts = List.of("--bootstrap-server", cluster.bootstrapServers(), "--entity-type", "topics", "--alter");
        try (Admin client = cluster.admin()) {
            client.createTopics(List.of(new NewTopic("test-config-topic", 1, (short) 1))).all().get();
            assertInstanceOf(
                    InvalidConfigurationException.class,
                    assertThrows(
                            ExecutionException.class,
                            () -> ConfigCommand.alterConfig(
                                    client,
                                    new ConfigCommand.ConfigCommandOptions(
                                            toArray(alterOpts,
                                                    List.of("--add-config", "invalid=2", "--entity-type", "topics", "--entity-name", "test-config-topic"))))
                    ).getCause()
            );
        }
    }

    // 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"),