content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.tools;
  
  import kafka.admin.ConfigCommand;
  
  import org.apache.kafka.clients.admin.Admin;
  import org.apache.kafka.clients.admin.AdminClientTestUtils;
  import org.apache.kafka.clients.admin.AlterConfigsOptions;
  import org.apache.kafka.clients.admin.AlterConfigsResult;
  import org.apache.kafka.clients.admin.ConfigEntry;
  import org.apache.kafka.clients.admin.NewTopic;
  import org.apache.kafka.clients.producer.Producer;
  import org.apache.kafka.common.config.ConfigResource;
  import org.apache.kafka.common.errors.InvalidConfigurationException;
  import org.apache.kafka.common.errors.UnsupportedVersionException;
  import org.apache.kafka.common.test.ClusterInstance;
  import org.apache.kafka.common.test.api.ClusterConfigProperty;
  import org.apache.kafka.common.test.api.ClusterTest;
  import org.apache.kafka.common.test.api.Type;
  import org.apache.kafka.common.utils.Exit;
  import org.apache.kafka.server.common.MetadataVersion;
  import org.apache.kafka.test.TestUtils;
  
  import org.mockito.Mockito;
  
  import java.time.Duration;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.atomic.AtomicReference;
  import java.util.function.Consumer;
  import java.util.stream.Collectors;
  import java.util.stream.Stream;
  
  import static java.util.Arrays.asList;
  import static java.util.Collections.singleton;
  import static java.util.Collections.singletonList;
  import static java.util.Collections.singletonMap;
  import static org.apache.kafka.common.config.SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG;
  import static org.apache.kafka.common.config.SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG;
  import static org.apache.kafka.common.config.SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG;
  import static org.apache.kafka.coordinator.group.GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG;
  import static org.apache.kafka.coordinator.group.GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG;
  import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG;
  import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG;
  import static org.apache.kafka.server.config.ReplicationConfigs.AUTO_LEADER_REBALANCE_ENABLE_CONFIG;
  import static org.apache.kafka.server.config.ServerConfigs.MESSAGE_MAX_BYTES_CONFIG;
  import static org.apache.kafka.server.config.ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG;
  import static org.apache.kafka.tools.ToolsTestUtils.captureStandardErr;
  import static org.apache.kafka.tools.ToolsTestUtils.captureStandardOut;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertInstanceOf;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.ArgumentMatchers.anyMap;
  
  public class ConfigCommandIntegrationTest {
      private final String defaultBrokerId = "0";
      private final String defaultGroupName = "group";
      private final String defaultClientMetricsName = "cm";
      private final ClusterInstance cluster;
  
      private static Runnable run(Stream<String> command) {
          return () -> {
              try {
                  ConfigCommand.main(command.toArray(String[]::new));
              } catch (RuntimeException e) {
                  // do nothing.
              } finally {
                  Exit.resetExitProcedure();
              }
          };
      }
  
      public ConfigCommandIntegrationTest(ClusterInstance cluster) {
          this.cluster = cluster;
      }
  
      @ClusterTest
      public void testExitWithNonZeroStatusOnUpdatingUnallowedConfig() {
          assertNonZeroStatusExit(Stream.concat(quorumArgs(), Stream.of(
              "--entity-name", "0",
              "--entity-type", "brokers",
              "--alter",
              "--add-config", "security.inter.broker.protocol=PLAINTEXT")),
              errOut -> assertTrue(errOut.contains("Cannot update these configs dynamically: Set(security.inter.broker.protocol)"), errOut));
      }
  
      @ClusterTest
      public void testNullStatusOnKraftCommandAlterUserQuota() {
          Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "users",
              "--entity-name", "admin",
              "--alter", "--add-config", "consumer_byte_rate=20000"));
          String message = captureStandardOut(run(command));
          assertEquals("Completed updating config for user admin.", message);
      }
  
      @ClusterTest
      public void testNullStatusOnKraftCommandAlterGroup() {
          Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "groups",
              "--entity-name", "group",
              "--alter", "--add-config", "consumer.session.timeout.ms=50000"));
          String message = captureStandardOut(run(command));
          assertEquals("Completed updating config for group group.", message);
  
          // Test for the --group alias
          command = Stream.concat(quorumArgs(), Stream.of(
              "--group", "group",
              "--alter", "--add-config", "consumer.session.timeout.ms=50000"));
          message = captureStandardOut(run(command));
          assertEquals("Completed updating config for group group.", message);
  
          // A non-existent group with dynamic configs can be described
          command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "groups",
              "--describe"));
          message = captureStandardOut(run(command));
          assertTrue(message.contains("Dynamic configs for group group are:"));
          assertTrue(message.contains("consumer.session.timeout.ms=50000 sensitive=false synonyms={DYNAMIC_GROUP_CONFIG:consumer.session.timeout.ms=50000}"));
  
          command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "groups",
              "--entity-name", "group",
              "--describe"));
          message = captureStandardOut(run(command));
          assertTrue(message.contains("Dynamic configs for group group are:"));
          assertTrue(message.contains("consumer.session.timeout.ms=50000 sensitive=false synonyms={DYNAMIC_GROUP_CONFIG:consumer.session.timeout.ms=50000}"));
      }
  
      @ClusterTest(serverProperties = {
          @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
          @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1")
      })
      public void testDescribeGroupWithoutDynamicConfigs(ClusterInstance cluster) throws InterruptedException, ExecutionException {
          cluster.createTopic("topic", 1, (short) 1);
  
          try (Producer<byte[], byte[]> producer = cluster.producer();
               org.apache.kafka.clients.consumer.Consumer<byte[], byte[]> consumer = cluster.consumer(Map.of(
                   "group.protocol", "consumer",
                   "group.id", "group"
               ))) {
              producer.send(new org.apache.kafka.clients.producer.ProducerRecord<>("topic", "key".getBytes(), "value".getBytes())).get();
              producer.flush();
              consumer.subscribe(List.of("topic"));
              consumer.poll(Duration.ofMillis(100));
  
              TestUtils.waitForCondition(() -> {
                  Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
                      "--entity-type", "groups",
                      "--describe"));
                  String message = captureStandardOut(run(command));
                  return message.contains("Dynamic configs for group group are:");
              }, () -> "cannot describe group without dynamic groups");
  
              TestUtils.waitForCondition(() -> {
                  Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
                      "--entity-type", "groups",
                      "--entity-name", "group",
                      "--describe"));
                  String message = captureStandardOut(run(command));
                  return message.contains("Dynamic configs for group group are:");
              }, () -> "cannot describe group without dynamic groups");
          }
      }
  
      @ClusterTest
      public void testNullStatusOnKraftCommandAlterClientMetrics() {
          Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
                  "--entity-type", "client-metrics",
                  "--entity-name", "cm",
                  "--alter", "--add-config", "metrics=org.apache"));
          String message = captureStandardOut(run(command));
          assertEquals("Completed updating config for client-metric cm.", message);
  
          // Test for the --client-metrics alias
          command = Stream.concat(quorumArgs(), Stream.of(
                  "--client-metrics", "cm",
                  "--alter", "--add-config", "metrics=org.apache"));
          message = captureStandardOut(run(command));
          assertEquals("Completed updating config for client-metric cm.", message);
  
          command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "client-metrics",
              "--describe"));
          message = captureStandardOut(run(command));
          assertTrue(message.contains("Dynamic configs for client-metric cm are:"));
          assertTrue(message.contains("metrics=org.apache sensitive=false synonyms={DYNAMIC_CLIENT_METRICS_CONFIG:metrics=org.apache}"));
  
          command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "client-metrics",
              "--entity-name", "cm",
              "--describe"));
          message = captureStandardOut(run(command));
          assertTrue(message.contains("Dynamic configs for client-metric cm are:"));
          assertTrue(message.contains("metrics=org.apache sensitive=false synonyms={DYNAMIC_CLIENT_METRICS_CONFIG:metrics=org.apache}"));
      }
  
      @ClusterTest
      public void testAddConfigKeyValuesUsingCommand() throws Exception {
          try (Admin client = cluster.admin()) {
              NewTopic newTopic = new NewTopic("topic", 1, (short) 1);
              client.createTopics(Set.of(newTopic)).all().get();
              cluster.waitForTopic("topic", 1);
              Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
                      "--entity-type", "topics",
                      "--entity-name", "topic",
                      "--alter", "--add-config", "cleanup.policy=[delete,compact]"));
              String message = captureStandardOut(run(command));
              assertEquals("Completed updating config for topic topic.", message);
              command = Stream.concat(quorumArgs(), Stream.of(
                      "--entity-type", "topics",
                      "--entity-name", "topic",
                      "--describe"));
              message = captureStandardOut(run(command));
              assertTrue(message.contains("cleanup.policy=delete,compact"), "Config entry was not added correctly");
          }
      }
  
      @ClusterTest
      public void testDynamicBrokerConfigUpdateUsingKraft() throws Exception {
          List<String> alterOpts = generateDefaultAlterOpts(cluster.bootstrapServers());
  
          try (Admin client = cluster.admin()) {
              // Add config
              alterAndVerifyConfig(client, Optional.of(defaultBrokerId), singletonMap(MESSAGE_MAX_BYTES_CONFIG, "110000"), alterOpts);
              alterAndVerifyConfig(client, Optional.empty(), singletonMap(MESSAGE_MAX_BYTES_CONFIG, "120000"), alterOpts);
  
              // Change config
              alterAndVerifyConfig(client, Optional.of(defaultBrokerId), singletonMap(MESSAGE_MAX_BYTES_CONFIG, "130000"), alterOpts);
              alterAndVerifyConfig(client, Optional.empty(), singletonMap(MESSAGE_MAX_BYTES_CONFIG, "140000"), alterOpts);
  
              // Delete config
              deleteAndVerifyConfigValue(client, defaultBrokerId, singleton(MESSAGE_MAX_BYTES_CONFIG), true, alterOpts);
  
              // Listener configs: should work only with listener name
              alterAndVerifyConfig(client, Optional.of(defaultBrokerId),
                      singletonMap("listener.name.internal.ssl.keystore.location", "/tmp/test.jks"), alterOpts);
              // Per-broker config configured at default cluster-level should fail
              assertThrows(ExecutionException.class,
                      () -> alterConfigWithAdmin(client, Optional.empty(),
                              singletonMap("listener.name.internal.ssl.keystore.location", "/tmp/test.jks"), alterOpts));
              deleteAndVerifyConfigValue(client, defaultBrokerId,
                      singleton("listener.name.internal.ssl.keystore.location"), false, alterOpts);
              alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                      singletonMap("listener.name.external.ssl.keystore.password", "secret"), alterOpts);
  
              Map<String, String> configs = new HashMap<>();
              configs.put("listener.name.external.ssl.keystore.password", "secret");
              configs.put("log.cleaner.threads", "2");
  
              // Password config update at default cluster-level should fail
              assertThrows(ExecutionException.class,
                      () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId), configs, alterOpts));
          }
      }
  
      @ClusterTest
      public void testGroupConfigUpdateUsingKraft() throws Exception {
          List<String> alterOpts = Stream.concat(entityOp(Optional.of(defaultGroupName)).stream(),
                          Stream.of("--entity-type", "groups", "--alter"))
                  .collect(Collectors.toList());
          verifyGroupConfigUpdate(alterOpts);
  
          // Test for the --group alias
          verifyGroupConfigUpdate(asList("--group", defaultGroupName, "--alter"));
      }
  
      @ClusterTest
      public void testDescribeStreamsGroupConfigs() {
          Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "groups",
              "--entity-name", "group",
              "--describe", "--all"));
          String message = captureStandardOut(run(command));
  
          assertTrue(message.contains("streams.heartbeat.interval.ms=5000 sensitive=false synonyms={DEFAULT_CONFIG:streams.heartbeat.interval.ms=5000}"));
          assertTrue(message.contains("streams.num.standby.replicas=0 sensitive=false synonyms={DEFAULT_CONFIG:streams.num.standby.replicas=0}"));
          assertTrue(message.contains("streams.session.timeout.ms=45000 sensitive=false synonyms={DEFAULT_CONFIG:streams.session.timeout.ms=45000}"));
      }
  
      @ClusterTest
      public void testAlterStreamsGroupNumOfStandbyReplicas() {
          // Verify the initial config
          Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "groups",
              "--entity-name", "group",
              "--describe", "--all"));
          String message = captureStandardOut(run(command));
          assertTrue(message.contains("streams.num.standby.replicas=0"));
  
          // Alter number of standby replicas
          command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "groups",
              "--entity-name", "group",
              "--alter", "--add-config", "streams.num.standby.replicas=1"));
          message = captureStandardOut(run(command));
          assertEquals("Completed updating config for group group.", message);
  
          // Verify the updated config
          command = Stream.concat(quorumArgs(), Stream.of(
              "--entity-type", "groups",
              "--entity-name", "group",
              "--describe"));
          message = captureStandardOut(run(command));
          assertTrue(message.contains("streams.num.standby.replicas=1"));
      }
  
      private void verifyGroupConfigUpdate(List<String> alterOpts) throws Exception {
          try (Admin client = cluster.admin()) {
              // Add config
              Map<String, String> configs = new HashMap<>();
              configs.put(CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "50000");
              configs.put(CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "6000");
              alterAndVerifyGroupConfig(client, defaultGroupName, configs, alterOpts);
  
              // Delete config
              configs.put(CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "45000");
              configs.put(CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "5000");
              deleteAndVerifyGroupConfigValue(client, defaultGroupName, configs, alterOpts);
  
              // Unknown config configured should fail
              assertThrows(ExecutionException.class, () -> alterConfigWithAdmin(client, singletonMap("unknown.config", "20000"), alterOpts));
          }
      }
  
  
      @ClusterTest(types = {Type.KRAFT})
      public void testClientMetricsConfigUpdate() throws Exception {
          List<String> alterOpts = Stream.concat(entityOp(Optional.of(defaultClientMetricsName)).stream(),
                          Stream.of("--entity-type", "client-metrics", "--alter"))
              .collect(Collectors.toList());
          verifyClientMetricsConfigUpdate(alterOpts);
  
          // Test for the --client-metrics alias
          verifyClientMetricsConfigUpdate(asList("--client-metrics", defaultClientMetricsName, "--alter"));
      }
  
      private void verifyClientMetricsConfigUpdate(List<String> alterOpts) throws Exception {
          try (Admin client = cluster.admin()) {
              // Add config
              Map<String, String> configs = new HashMap<>();
              configs.put("metrics", "org.apache.kafka.producer.");
              configs.put("interval.ms", "6000");
              alterAndVerifyClientMetricsConfig(client, defaultClientMetricsName, configs, alterOpts);
  
              // Delete config
              configs.put("metrics", "");
              configs.put("interval.ms", "300000");
              deleteAndVerifyClientMetricsConfigValue(client, defaultClientMetricsName, configs, alterOpts);
  
              // Unknown config configured should fail
              assertThrows(ExecutionException.class, () -> alterConfigWithAdmin(client, singletonMap("unknown.config", "20000"), alterOpts));
          }
      }
  
      @ClusterTest
      public void testAlterReadOnlyConfigInKRaftThenShouldFail() {
          List<String> alterOpts = generateDefaultAlterOpts(cluster.bootstrapServers());
  
          try (Admin client = cluster.admin()) {
              assertThrows(ExecutionException.class,
                      () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                              singletonMap(AUTO_CREATE_TOPICS_ENABLE_CONFIG, "false"), alterOpts));
              assertThrows(ExecutionException.class,
                      () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                              singletonMap(AUTO_LEADER_REBALANCE_ENABLE_CONFIG, "false"), alterOpts));
              assertThrows(ExecutionException.class,
                      () -> alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                              singletonMap("broker.id", "1"), alterOpts));
          }
      }
  
      @ClusterTest
      public void testUpdateClusterWideConfigInKRaftThenShouldSuccessful() throws Exception {
          List<String> alterOpts = generateDefaultAlterOpts(cluster.bootstrapServers());
  
          try (Admin client = cluster.admin()) {
              alterAndVerifyConfig(client, Optional.of(defaultBrokerId),
                      singletonMap("log.flush.interval.messages", "100"), alterOpts);
              alterAndVerifyConfig(client, Optional.of(defaultBrokerId),
                      singletonMap("log.retention.bytes", "20"), alterOpts);
              alterAndVerifyConfig(client, Optional.of(defaultBrokerId),
                      singletonMap("log.retention.ms", "2"), alterOpts);
          }
      }
  
      @ClusterTest
      public void testUpdatePerBrokerConfigWithListenerNameInKRaftThenShouldSuccessful() throws Exception {
          List<String> alterOpts = generateDefaultAlterOpts(cluster.bootstrapServers());
          String listenerName = "listener.name.internal.";
  
          try (Admin client = cluster.admin()) {
              alterAndVerifyConfig(client, Optional.of(defaultBrokerId),
                      singletonMap(listenerName + "ssl.truststore.type", "PKCS12"), alterOpts);
              alterAndVerifyConfig(client, Optional.of(defaultBrokerId),
                      singletonMap(listenerName + "ssl.truststore.location", "/temp/test.jks"), alterOpts);
              alterConfigWithAdmin(client, Optional.of(defaultBrokerId),
                      singletonMap(listenerName + "ssl.truststore.password", "password"), alterOpts);
              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(cluster.anyBrokerSocketServer().config().brokerId() + ""));
      }
  
      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()) {
              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,
                                                      asList("--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"),
      })
      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.brokerSocketServers().stream().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(
                                  toArray(asList(
                                          "--bootstrap-server", cluster.bootstrapServers(),
                                          "--alter",
                                          "--add-config", "log.cleaner.threads=2",
                                          "--entity-type", "brokers",
                                          "--entity-default"))));
                      }).getMessage()
              );
              Mockito.verify(spyAdmin).incrementalAlterConfigs(anyMap(), any(AlterConfigsOptions.class));
          }
      }
  
      @ClusterTest
      public void testDescribeNonExistentConfigResource() {
          Map<String, String> configResourceTypeAndNames = Map.of(
              "brokers", "3",
              "broker-loggers", "3",
              "topics", "non-existent",
              "groups", "non-existent",
              "client-metrics", "non-existent");
          configResourceTypeAndNames.forEach((type, name) -> {
              Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
                  "--entity-type", type,
                  "--entity-name", name,
                  "--describe"));
              String message = captureStandardOut(run(command));
              assertTrue(
                  message.contains("The " + type.substring(0, type.length() - 1) + " '" + name + "' doesn't exist and doesn't have dynamic config."),
                  "The config resource type " + type + " got unexpected result: " + message
              );
          });
      }
  
      private void assertNonZeroStatusExit(Stream<String> args, Consumer<String> checkErrOut) {
          AtomicReference<Integer> exitStatus = new AtomicReference<>();
          Exit.setExitProcedure((status, __) -> {
              exitStatus.set(status);
              throw new RuntimeException();
          });
  
          String errOut = captureStandardErr(run(args));
  
          checkErrOut.accept(errOut);
          assertNotNull(exitStatus.get());
          assertEquals(1, exitStatus.get());
      }
  
      private Stream<String> quorumArgs() {
          return Stream.of("--bootstrap-server", cluster.bootstrapServers());
      }
  
      private List<String> entityOp(Optional<String> entityId) {
          return entityId.map(id -> asList("--entity-name", id))
                  .orElse(singletonList("--entity-default"));
      }
  
      private List<String> generateDefaultAlterOpts(String bootstrapServers) {
          return asList("--bootstrap-server", bootstrapServers,
                  "--entity-type", "brokers", "--alter");
      }
  
      private void alterAndVerifyConfig(Admin client,
                                        Optional<String> brokerId,
                                        Map<String, String> config,
                                        List<String> alterOpts) throws Exception {
          alterConfigWithAdmin(client, brokerId, config, alterOpts);
          verifyConfig(client, brokerId, config);
      }
  
      private void alterAndVerifyGroupConfig(Admin client,
                                             String groupName,
                                             Map<String, String> config,
                                             List<String> alterOpts) throws Exception {
          alterConfigWithAdmin(client, config, alterOpts);
          verifyGroupConfig(client, groupName, config);
      }
  
      private void alterAndVerifyClientMetricsConfig(Admin client,
                                                     String clientMetricsName,
                                                     Map<String, String> config,
                                                     List<String> alterOpts) throws Exception {
          alterConfigWithAdmin(client, config, alterOpts);
          verifyClientMetricsConfig(client, clientMetricsName, config);
      }
  
      private void alterConfigWithAdmin(Admin client, Optional<String> resourceName, Map<String, String> config, List<String> alterOpts) {
          String configStr = transferConfigMapToString(config);
          List<String> bootstrapOpts = quorumArgs().collect(Collectors.toList());
          ConfigCommand.ConfigCommandOptions addOpts =
                  new ConfigCommand.ConfigCommandOptions(toArray(bootstrapOpts,
                          entityOp(resourceName),
                          alterOpts,
                          asList("--add-config", configStr)));
          addOpts.checkArgs();
          ConfigCommand.alterConfig(client, addOpts);
      }
  
      private void alterConfigWithAdmin(Admin client, Map<String, String> config, List<String> alterOpts) {
          String configStr = transferConfigMapToString(config);
          List<String> bootstrapOpts = quorumArgs().collect(Collectors.toList());
          ConfigCommand.ConfigCommandOptions addOpts =
                  new ConfigCommand.ConfigCommandOptions(toArray(bootstrapOpts,
                          alterOpts,
                          asList("--add-config", configStr)));
          addOpts.checkArgs();
          ConfigCommand.alterConfig(client, addOpts);
      }
  
      private void verifyConfig(Admin client, Optional<String> brokerId, Map<String, String> config) throws Exception {
          ConfigResource configResource = new ConfigResource(ConfigResource.Type.BROKER, brokerId.orElse(""));
          TestUtils.waitForCondition(() -> {
              Map<String, String> current = getConfigEntryStream(client, configResource)
                      .filter(configEntry -> Objects.nonNull(configEntry.value()))
                      .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value));
              return config.entrySet().stream().allMatch(e -> e.getValue().equals(current.get(e.getKey())));
          }, 10000, config + " are not updated");
      }
  
      private void verifyGroupConfig(Admin client, String groupName, Map<String, String> config) throws Exception {
          ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupName);
          TestUtils.waitForCondition(() -> {
              Map<String, String> current = getConfigEntryStream(client, configResource)
                  .filter(configEntry -> Objects.nonNull(configEntry.value()))
                  .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value));
              return config.entrySet().stream().allMatch(e -> e.getValue().equals(current.get(e.getKey())));
          }, 10000, config + " are not updated");
      }
  
      private void verifyClientMetricsConfig(Admin client, String clientMetricsName, Map<String, String> config) throws Exception {
          ConfigResource configResource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, clientMetricsName);
          TestUtils.waitForCondition(() -> {
              Map<String, String> current = getConfigEntryStream(client, configResource)
                      .filter(configEntry -> Objects.nonNull(configEntry.value()))
                      .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value));
              if (config.isEmpty())
                  return current.isEmpty();
              return config.entrySet().stream().allMatch(e -> e.getValue().equals(current.get(e.getKey())));
          }, 10000, config + " are not updated");
      }
  
      private Stream<ConfigEntry> getConfigEntryStream(Admin client,
                                                       ConfigResource configResource) throws InterruptedException, ExecutionException {
          return client.describeConfigs(singletonList(configResource))
                  .all()
                  .get()
                  .values()
                  .stream()
                  .flatMap(e -> e.entries().stream());
      }
  
      private void deleteAndVerifyConfigValue(Admin client,
                                              String brokerId,
                                              Set<String> config,
                                              boolean hasDefaultValue,
                                              List<String> alterOpts) throws Exception {
          ConfigCommand.ConfigCommandOptions deleteOpts =
                  new ConfigCommand.ConfigCommandOptions(toArray(alterOpts, asList("--entity-name", brokerId),
                          asList("--delete-config", String.join(",", config))));
          deleteOpts.checkArgs();
          ConfigCommand.alterConfig(client, deleteOpts);
          verifyPerBrokerConfigValue(client, brokerId, config, hasDefaultValue);
      }
  
      private void deleteAndVerifyGroupConfigValue(Admin client,
                                                   String groupName,
                                                   Map<String, String> defaultConfigs,
                                                   List<String> alterOpts) throws Exception {
          List<String> bootstrapOpts = quorumArgs().collect(Collectors.toList());
          ConfigCommand.ConfigCommandOptions deleteOpts =
              new ConfigCommand.ConfigCommandOptions(toArray(bootstrapOpts,
                      alterOpts,
                      asList("--delete-config", String.join(",", defaultConfigs.keySet()))));
          deleteOpts.checkArgs();
          ConfigCommand.alterConfig(client, deleteOpts);
          verifyGroupConfig(client, groupName, defaultConfigs);
      }
  
      private void deleteAndVerifyClientMetricsConfigValue(Admin client,
                                                           String clientMetricsName,
                                                           Map<String, String> defaultConfigs,
                                                           List<String> alterOpts) throws Exception {
          List<String> bootstrapOpts = quorumArgs().collect(Collectors.toList());
          ConfigCommand.ConfigCommandOptions deleteOpts =
              new ConfigCommand.ConfigCommandOptions(toArray(bootstrapOpts,
                      alterOpts,
                      asList("--delete-config", String.join(",", defaultConfigs.keySet()))));
          deleteOpts.checkArgs();
          ConfigCommand.alterConfig(client, deleteOpts);
          verifyClientMetricsConfig(client, clientMetricsName, defaultConfigs);
      }
  
      private void verifyPerBrokerConfigValue(Admin client,
                                              String brokerId,
                                              Set<String> config,
                                              boolean hasDefaultValue) throws Exception {
          ConfigResource configResource = new ConfigResource(ConfigResource.Type.BROKER, brokerId);
          TestUtils.waitForCondition(() -> {
              if (hasDefaultValue) {
                  Map<String, String> current = getConfigEntryStream(client, configResource)
                          .filter(configEntry -> Objects.nonNull(configEntry.value()))
                          .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value));
                  return config.stream().allMatch(current::containsKey);
              } else {
                  return getConfigEntryStream(client, configResource)
                          .noneMatch(configEntry -> config.contains(configEntry.name()));
              }
          }, 5000, config + " are not updated");
      }
  
      private void verifyConfigSecretValue(Admin client, Optional<String> brokerId, Set<String> config) throws Exception {
          ConfigResource configResource = new ConfigResource(ConfigResource.Type.BROKER, brokerId.orElse(""));
          TestUtils.waitForCondition(() -> {
              Map<String, String> current = getConfigEntryStream(client, configResource)
                      .filter(ConfigEntry::isSensitive)
                      .collect(HashMap::new, (map, entry) -> map.put(entry.name(), entry.value()), HashMap::putAll);
              return config.stream().allMatch(current::containsKey);
          }, 5000, config + " are not updated");
      }
  
      @SafeVarargs
      private static String[] toArray(List<String>... lists) {
          return Stream.of(lists).flatMap(List::stream).toArray(String[]::new);
      }
  
      private String transferConfigMapToString(Map<String, String> configs) {
          return configs.entrySet()
                  .stream()
                  .map(e -> e.getKey() + "=" + e.getValue())
                  .collect(Collectors.joining(","));
      }
  }
  