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

Code before:
            verifyConfigSecretValue(client, Optional.of(defaultBrokerId),
                    singleton(listenerName + "ssl.truststore.password"));
        }
    }

    @ClusterTest
    public void testUpdatePerBrokerConfigInKRaftThenShouldFail() {
        List<String> alterOpts = generateDefaultAlterOpts(cluster.bootstrapServers());

        try (Admin client = cluster.admin()) {
            assertThrows(ExecutionException.class,
                    () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                            singletonMap(SSL_TRUSTSTORE_TYPE_CONFIG, "PKCS12"), alterOpts));
            assertThrows(ExecutionException.class,
                    () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                            singletonMap(SSL_TRUSTSTORE_LOCATION_CONFIG, "/temp/test.jks"), alterOpts));
            assertThrows(ExecutionException.class,
                    () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                            singletonMap(SSL_TRUSTSTORE_PASSWORD_CONFIG, "password"), alterOpts));
        }
    }

    @ClusterTest
    public void testUpdateInvalidBrokerConfigs() {
        updateAndCheckInvalidBrokerConfig(Optional.empty());
        updateAndCheckInvalidBrokerConfig(cluster.brokers().values().stream().map(broker -> broker.socketServer().config().brokerId() + "").findFirst());
    }

    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 = asList("--bootstrap-server", cluster.bootstrapServers(), "--entity-type", "topics", "--alter");
        try (Admin client = cluster.admin()) {

Review comment (suggestion):
```java
updateAndCheckInvalidBrokerConfig(Optional.of(String.valueOf((cluster.brokers().entrySet().iterator().next().getKey()))));
```

Code after:
            verifyConfigSecretValue(client, Optional.of(defaultBrokerId),
                    singleton(listenerName + "ssl.truststore.password"));
        }
    }

    @ClusterTest
    public void testUpdatePerBrokerConfigInKRaftThenShouldFail() {
        List<String> alterOpts = generateDefaultAlterOpts(cluster.bootstrapServers());

        try (Admin client = cluster.admin()) {
            assertThrows(ExecutionException.class,
                    () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                            singletonMap(SSL_TRUSTSTORE_TYPE_CONFIG, "PKCS12"), alterOpts));
            assertThrows(ExecutionException.class,
                    () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                            singletonMap(SSL_TRUSTSTORE_LOCATION_CONFIG, "/temp/test.jks"), alterOpts));
            assertThrows(ExecutionException.class,
                    () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                            singletonMap(SSL_TRUSTSTORE_PASSWORD_CONFIG, "password"), alterOpts));
        }
    }

    @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 = asList("--bootstrap-server", cluster.bootstrapServers(), "--entity-type", "topics", "--alter");
        try (Admin client = cluster.admin()) {