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

Code before:
    public void testUpdateInvalidBrokerConfigs() throws InterruptedException {
        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);

                boolean hasInvalidNull = describeResult.contains("invalid=null");
                boolean hasSensitiveTrue = describeResult.contains("sensitive=true");

                if (hasInvalidNull && !hasSensitiveTrue) {
                    throw new AssertionError("Expected sensitive=true when invalid=null is visible, but it was not present.\n" +
                            "Describe output:\n" + describeResult);
                }

                return hasSensitiveTrue && hasInvalidNull;
            }, 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()
            );
        }

Review comment (security):
will this truly make it interrupt the `waitForCondition`? or is it maybe going to continue retrying? (from a quick look at the code path seems it ends up using `retryOnExceptionWithTimeout`?)

Maybe we can just `waitForCondition` checking `describeResult.contains("invalid=null") `, and then, outside of the wait, assert that `describeResult.contains("sensitive=true");` wdyt? 

nit: the message seems a bit confusing to me, maybe something along the lines of "Description of the invalid config did not include sensive=true as expected"

Code after:
    public void testUpdateInvalidBrokerConfigs() throws InterruptedException {
        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.