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.clients.admin;
  
  
  import kafka.server.KafkaConfig;
  
  import org.apache.kafka.common.config.ConfigDef;
  import org.apache.kafka.common.config.ConfigResource;
  import org.apache.kafka.common.config.TopicConfig;
  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.coordinator.group.GroupConfig;
  import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
  import org.apache.kafka.server.config.KRaftConfigs;
  import org.apache.kafka.server.config.ServerConfigs;
  import org.apache.kafka.server.metrics.ClientMetricsConfigs;
  import org.apache.kafka.storage.internals.log.LogConfig;
  
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.ExecutionException;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertNotEquals;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  
  public class StaticBrokerConfigTest {
      private static final String TOPIC = "topic";
      private static final String CUSTOM_VALUE = "1048576";
  
      /**
       * synonyms of `segment.bytes`
       */
      private static final String LOG_SEGMENT_BYTES = "log.segment.bytes";
  
      @ClusterTest(types = {Type.KRAFT},
          serverProperties = {
              @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
              @ClusterConfigProperty(id = 3000, key = LOG_SEGMENT_BYTES, value = CUSTOM_VALUE)
          })
      public void testTopicConfigsGetImpactedIfStaticConfigsAddToController(ClusterInstance cluster)
          throws ExecutionException, InterruptedException {
          try (
              Admin admin = cluster.admin();
              Admin adminUsingBootstrapController = cluster.admin(Map.of(), true)
          ) {
              ConfigEntry configEntry = admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) 1)))
                  .config(TOPIC).get().get(TopicConfig.SEGMENT_BYTES_CONFIG);
              assertEquals(ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG, configEntry.source());
              assertEquals(CUSTOM_VALUE, configEntry.value(), "Config value should be custom value since controller has related static config");
  
              ConfigResource brokerResource = new ConfigResource(ConfigResource.Type.BROKER, "0");
              configEntry = admin.describeConfigs(List.of(brokerResource)).all().get().get(brokerResource).get(LOG_SEGMENT_BYTES);
              assertEquals(ConfigEntry.ConfigSource.DEFAULT_CONFIG, configEntry.source());
              assertNotEquals(CUSTOM_VALUE,
                  configEntry.value(),
                  "Config value should not be custom value since broker doesn't have related static config");
  
              ConfigResource controllerResource = new ConfigResource(ConfigResource.Type.BROKER, "3000");
              configEntry = adminUsingBootstrapController.describeConfigs(List.of(controllerResource))
                  .all().get().get(controllerResource).get(LOG_SEGMENT_BYTES);
              assertEquals(ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG, configEntry.source());
              assertEquals(CUSTOM_VALUE,
                  configEntry.value(),
                  "Config value should be custom value since controller has related static config");
          }
      }
  
      @ClusterTest(types = {Type.KRAFT},
          serverProperties = {
              @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
              @ClusterConfigProperty(id = 0, key = LOG_SEGMENT_BYTES, value = CUSTOM_VALUE)
          })
      public void testTopicConfigsGetImpactedIfStaticConfigsAddToBroker(ClusterInstance cluster)
          throws ExecutionException, InterruptedException {
          try (
              Admin admin = cluster.admin();
              Admin adminUsingBootstrapController = cluster.admin(Map.of(), true)
          ) {
              ConfigEntry configEntry = admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) 1)))
                  .config(TOPIC).get().get(TopicConfig.SEGMENT_BYTES_CONFIG);
              assertEquals(ConfigEntry.ConfigSource.DEFAULT_CONFIG, configEntry.source());
              assertNotEquals(CUSTOM_VALUE,
                  configEntry.value(),
                  "Config value should not be custom value since controller doesn't have static config");
  
              ConfigResource brokerResource = new ConfigResource(ConfigResource.Type.BROKER, "0");
              configEntry = admin.describeConfigs(List.of(brokerResource)).all().get().get(brokerResource).get(LOG_SEGMENT_BYTES);
              assertEquals(ConfigEntry.ConfigSource.STATIC_BROKER_CONFIG, configEntry.source());
              assertEquals(CUSTOM_VALUE,
                  configEntry.value(),
                  "Config value should be custom value since broker has related static config");
  
              ConfigResource controllerResource = new ConfigResource(ConfigResource.Type.BROKER, "3000");
              configEntry = adminUsingBootstrapController.describeConfigs(List.of(controllerResource))
                  .all().get().get(controllerResource).get(LOG_SEGMENT_BYTES);
              assertEquals(ConfigEntry.ConfigSource.DEFAULT_CONFIG, configEntry.source());
              assertNotEquals(CUSTOM_VALUE,
                  configEntry.value(),
                  "Config value should not be custom value since controller doesn't have related static config");
          }
      }
  
      @ClusterTest(types = {Type.KRAFT})
      public void testInternalConfigsDoNotReturnForDescribeConfigs(ClusterInstance cluster) throws Exception {
          try (
                  Admin admin = cluster.admin();
                  Admin controllerAdmin = cluster.admin(Map.of(), true)
          ) {
              ConfigResource brokerResource = new ConfigResource(ConfigResource.Type.BROKER, "0");
              ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC);
              ConfigResource groupResource = new ConfigResource(ConfigResource.Type.GROUP, "testGroup");
              ConfigResource clientMetricsResource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, "testClient");
  
              admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) 1))).config(TOPIC).get();
              // make sure the topic metadata exist
              cluster.waitForTopic(TOPIC, 1);
              Map<ConfigResource, Config> configResourceMap = admin.describeConfigs(
                      List.of(brokerResource, topicResource, groupResource, clientMetricsResource)).all().get();
  
              // test for case ConfigResource.Type == BROKER
              // Notice: since the testing framework actively sets three internal configurations when starting the
              // broker (see org.apache.kafka.common.test.KafkaClusterTestKit.Builder.createNodeConfig()),
              // so the API `describeConfigs` will also return these three configurations. However, other internal
              // configurations will not be returned
              Set<String> ignoreConfigNames = Set.of(
                      ServerConfigs.UNSTABLE_FEATURE_VERSIONS_ENABLE_CONFIG,
                      ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG,
                      KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG);
              Config brokerConfig = configResourceMap.get(brokerResource);
              assertNotContainsInternalConfig(brokerConfig, KafkaConfig.configDef().configKeys(), ignoreConfigNames);
  
              // test for case ConfigResource.Type == TOPIC
              Config topicConfig = configResourceMap.get(topicResource);
              assertNotContainsAnyInternalConfig(topicConfig, LogConfig.configKeys());
  
              // test for case ConfigResource.Type == GROUP
              Config groupConfig = configResourceMap.get(groupResource);
              assertNotContainsAnyInternalConfig(groupConfig, GroupConfig.configDef().configKeys());
  
              // test for case ConfigResource.Type == CLIENT_METRICS
              Config clientMetricsConfig = configResourceMap.get(clientMetricsResource);
              assertNotContainsAnyInternalConfig(clientMetricsConfig, ClientMetricsConfigs.configDef().configKeys());
  
              // test for controller node, and ConfigResource.Type == BROKER
              ConfigResource controllerResource = new ConfigResource(ConfigResource.Type.BROKER, "3000");
              Map<ConfigResource, Config> controllerConfigMap = controllerAdmin.describeConfigs(List.of(controllerResource)).all().get();
              Config controllerConfig = controllerConfigMap.get(controllerResource);
              assertNotContainsInternalConfig(controllerConfig, KafkaConfig.configDef().configKeys(), ignoreConfigNames);
          }
      }
  
      @ClusterTest(types = {Type.KRAFT})
      public void testInternalConfigsDoNotReturnForCreateTopics(ClusterInstance cluster) throws Exception {
          try (Admin admin = cluster.admin()) {
              // test for createTopics API
              Config config = admin.createTopics(List.of(new NewTopic(TOPIC, 1, (short) 1))).config(TOPIC).get();
              assertNotContainsAnyInternalConfig(config, LogConfig.configKeys());
          }
      }
  
      private void assertNotContainsAnyInternalConfig(Config config, Map<String, ConfigDef.ConfigKey> configKeyMap) {
          assertNotContainsInternalConfig(config, configKeyMap, Set.of());
      }
  
      private void assertNotContainsInternalConfig(Config config, Map<String, ConfigDef.ConfigKey> configKeyMap,
                                                   Set<String> ignoreConfigNames) {
          assertFalse(config.entries().isEmpty());
          for (ConfigEntry topicConfigEntry : config.entries()) {
              String configName = topicConfigEntry.name();
              ConfigDef.ConfigKey configKey = configKeyMap.get(configName);
  
              assertNotNull(configKey, "The ConfigKey of the config named '" + configName + "' should not be null");
              if (!ignoreConfigNames.contains(configName)) {
                  assertFalse(configKey.internalConfig, "The config named '" + configName + "' is an internal config and should not be returned");
              }
          }
      }
  }
  