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

Code before:
    }

    @ClusterTest
    public void testUpdateInvalidBrokerConfigs() {
        updateAndCheckInvalidBrokerConfig(Optional.empty());
        updateAndCheckInvalidBrokerConfig(Optional.of(String.valueOf((cluster.brokers().entrySet().iterator().next().getKey()))));
    }

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

            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));

            // We will treat unknown config as sensitive
            assertTrue(describeResult.contains("sensitive=true"), describeResult);
            // Sensitive config will not return
            assertTrue(describeResult.contains("invalid=null"), describeResult);
        }
    }

    @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"),
    })

Review comment (suggestion):
we need to update this msg after the latest changes right? (remove  the "missing 'invalid=null'"?)

Code after:
    }

    @ClusterTest
    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()
            );
        }
    }