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.coordinator.group;
  
  import org.apache.kafka.common.config.ConfigException;
  import org.apache.kafka.common.errors.InvalidConfigurationException;
  import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfig;
  import org.apache.kafka.coordinator.group.modern.share.ShareGroupConfigTest;
  
  import org.junit.jupiter.api.Test;
  
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Properties;
  
  import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  
  public class GroupConfigTest {
  
      private static final int OFFSET_METADATA_MAX_SIZE = 4096;
      private static final long OFFSETS_RETENTION_CHECK_INTERVAL_MS = 1000L;
      private static final int OFFSETS_RETENTION_MINUTES = 24 * 60;
  
      private static final boolean SHARE_GROUP_ENABLE = true;
      private static final int SHARE_GROUP_PARTITION_MAX_RECORD_LOCKS = 200;
      private static final int SHARE_GROUP_DELIVERY_COUNT_LIMIT = 5;
      private static final int SHARE_GROUP_RECORD_LOCK_DURATION_MS = 30000;
      private static final int SHARE_GROUP_MIN_RECORD_LOCK_DURATION_MS = 15000;
      private static final int SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS = 60000;
  
      @Test
      public void testFromPropsInvalid() {
          GroupConfig.configNames().forEach(name -> {
              if (GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "not_a_number", "-0.1", "1.2");
              } else if (GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "not_a_number", "-0.1", "1.2");
              } else if (GroupConfig.SHARE_SESSION_TIMEOUT_MS_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "not_a_number", "-0.1", "1.2");
              } else if (GroupConfig.SHARE_HEARTBEAT_INTERVAL_MS_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "not_a_number", "-0.1", "1.2");
              } else if (GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "not_a_number", "-0.1", "1.2");
              } else if (GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "hello", "1.0");
              } else if (GroupConfig.SHARE_ISOLATION_LEVEL_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "hello", "1.0");
              } else if (GroupConfig.STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "not_a_number", "1.0");
              } else if (GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "not_a_number", "1.0");
              } else if (GroupConfig.STREAMS_SESSION_TIMEOUT_MS_CONFIG.equals(name)) {
                  assertPropertyInvalid(name, "not_a_number", "1.0");
              } else {
                  assertPropertyInvalid(name, "not_a_number", "-0.1");
              }
          });
      }
  
      private void assertPropertyInvalid(String name, Object... values) {
          for (Object value : values) {
              Properties props = new Properties();
              props.setProperty(name, value.toString());
              assertThrows(Exception.class, () -> new GroupConfig(props));
          }
      }
  
      @Test
      public void testValidShareAutoOffsetResetValues() {
  
          Properties props = createValidGroupConfig();
  
          // Check for value "latest"
          props.put(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, "latest");
          doTestValidProps(props);
          props = createValidGroupConfig();
  
          // Check for value "earliest"
          props.put(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, "earliest");
          doTestValidProps(props);
  
          // Check for value "by_duration"
          props.put(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, "by_duration:PT10S");
          doTestValidProps(props);
      }
  
      @Test
      public void testValidShareIsolationLevelValues() {
          // Check for value READ_UNCOMMITTED
          Properties props = createValidGroupConfig();
          props.put(GroupConfig.SHARE_ISOLATION_LEVEL_CONFIG, "read_committed");
          doTestValidProps(props);
  
          // Check for value READ_COMMITTED
          props = createValidGroupConfig();
          props.put(GroupConfig.SHARE_ISOLATION_LEVEL_CONFIG, "read_uncommitted");
          doTestValidProps(props);
      }
  
      @Test
      public void testInvalidProps() {
  
          Properties props = createValidGroupConfig();
  
          // Check for invalid consumerSessionTimeoutMs, < MIN
          props.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "1");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid consumerSessionTimeoutMs, > MAX
          props.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "70000");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid consumerHeartbeatIntervalMs, < MIN
          props.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "1");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid consumerHeartbeatIntervalMs, > MAX
          props.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "70000");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid shareSessionTimeoutMs, < MIN
          props.put(GroupConfig.SHARE_SESSION_TIMEOUT_MS_CONFIG, "1");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid shareSessionTimeoutMs, > MAX
          props.put(GroupConfig.SHARE_SESSION_TIMEOUT_MS_CONFIG, "70000");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid shareHeartbeatIntervalMs, < MIN
          props.put(GroupConfig.SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, "1");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid shareHeartbeatIntervalMs, > MAX
          props.put(GroupConfig.SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, "70000");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid shareRecordLockDurationMs, < MIN
          props.put(GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG, "10000");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid shareRecordLockDurationMs, > MAX
          props.put(GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG, "70000");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid shareAutoOffsetReset
          props.put(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, "hello");
          doTestInvalidProps(props, ConfigException.class);
  
          // Check for invalid shareAutoOffsetReset, by_duration without duration
          props.put(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, "by_duration");
          doTestInvalidProps(props, ConfigException.class);
  
          // Check for invalid shareAutoOffsetReset, by_duration with negative duration
          props.put(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, "by_duration:-PT10S");
          doTestInvalidProps(props, ConfigException.class);
  
          // Check for invalid shareAutoOffsetReset, by_duration with invalid duration
          props.put(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, "by_duration:invalid");
          doTestInvalidProps(props, ConfigException.class);
          props = createValidGroupConfig();
  
          // Check for invalid streamsSessionTimeoutMs, < MIN
          props.put(GroupConfig.STREAMS_SESSION_TIMEOUT_MS_CONFIG, "1");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid streamsSessionTimeoutMs, > MAX
          props.put(GroupConfig.STREAMS_SESSION_TIMEOUT_MS_CONFIG, "70000");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid streamsHeartbeatIntervalMs, < MIN
          props.put(GroupConfig.STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG, "1000");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid streamsHeartbeatIntervalMs, > MAX
          props.put(GroupConfig.STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG, "70000");
          doTestInvalidProps(props, InvalidConfigurationException.class);
          props = createValidGroupConfig();
  
          // Check for invalid shareIsolationLevel.
          props.put(GroupConfig.SHARE_ISOLATION_LEVEL_CONFIG, "read_commit");
          doTestInvalidProps(props, ConfigException.class);
          props = createValidGroupConfig();
  
          // Check for invalid shareIsolationLevel.
          props.put(GroupConfig.SHARE_ISOLATION_LEVEL_CONFIG, "read_uncommit");
          doTestInvalidProps(props, ConfigException.class);
      }
  
      private void doTestInvalidProps(Properties props, Class<? extends Exception> exceptionClassName) {
          assertThrows(exceptionClassName, () -> GroupConfig.validate(props, createGroupCoordinatorConfig(), createShareGroupConfig()));
      }
  
      private void doTestValidProps(Properties props) {
          assertDoesNotThrow(() -> GroupConfig.validate(props, createGroupCoordinatorConfig(), createShareGroupConfig()));
      }
  
      @Test
      public void testFromPropsWithDefaultValue() {
          Map<String, String> defaultValue = new HashMap<>();
          defaultValue.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "10");
          defaultValue.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "10");
          defaultValue.put(GroupConfig.SHARE_SESSION_TIMEOUT_MS_CONFIG, "10");
          defaultValue.put(GroupConfig.SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, "10");
          defaultValue.put(GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG, "2000");
          defaultValue.put(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, "latest");
          defaultValue.put(GroupConfig.SHARE_ISOLATION_LEVEL_CONFIG, "read_uncommitted");
          defaultValue.put(GroupConfig.STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG, "10");
          defaultValue.put(GroupConfig.STREAMS_SESSION_TIMEOUT_MS_CONFIG, "2000");
          defaultValue.put(GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG, "1");
  
          Properties props = new Properties();
          props.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "20");
          GroupConfig config = GroupConfig.fromProps(defaultValue, props);
  
          assertEquals(10, config.getInt(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG));
          assertEquals(20, config.getInt(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG));
          assertEquals(10, config.getInt(GroupConfig.SHARE_HEARTBEAT_INTERVAL_MS_CONFIG));
          assertEquals(10, config.getInt(GroupConfig.SHARE_SESSION_TIMEOUT_MS_CONFIG));
          assertEquals(2000, config.getInt(GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG));
          assertEquals("latest", config.getString(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG));
          assertEquals("read_uncommitted", config.getString(GroupConfig.SHARE_ISOLATION_LEVEL_CONFIG));
          assertEquals(10, config.getInt(GroupConfig.STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG));
          assertEquals(2000, config.getInt(GroupConfig.STREAMS_SESSION_TIMEOUT_MS_CONFIG));
          assertEquals(1, config.getInt(GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG));
      }
  
      @Test
      public void testInvalidConfigName() {
          Properties props = new Properties();
          props.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "10");
          props.put("invalid.config.name", "10");
          assertThrows(InvalidConfigurationException.class, () -> GroupConfig.validate(props, createGroupCoordinatorConfig(), createShareGroupConfig()));
      }
  
      private Properties createValidGroupConfig() {
          Properties props = new Properties();
          props.put(GroupConfig.CONSUMER_SESSION_TIMEOUT_MS_CONFIG, "45000");
          props.put(GroupConfig.CONSUMER_HEARTBEAT_INTERVAL_MS_CONFIG, "5000");
          props.put(GroupConfig.SHARE_SESSION_TIMEOUT_MS_CONFIG, "45000");
          props.put(GroupConfig.SHARE_HEARTBEAT_INTERVAL_MS_CONFIG, "5000");
          props.put(GroupConfig.SHARE_RECORD_LOCK_DURATION_MS_CONFIG, "30000");
          props.put(GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, "latest");
          props.put(GroupConfig.SHARE_ISOLATION_LEVEL_CONFIG, "read_uncommitted");
          props.put(GroupConfig.STREAMS_SESSION_TIMEOUT_MS_CONFIG, "50000");
          props.put(GroupConfig.STREAMS_HEARTBEAT_INTERVAL_MS_CONFIG, "6000");
          props.put(GroupConfig.STREAMS_NUM_STANDBY_REPLICAS_CONFIG, "1");
          return props;
      }
  
      private GroupCoordinatorConfig createGroupCoordinatorConfig() {
          return GroupCoordinatorConfigTest.createGroupCoordinatorConfig(OFFSET_METADATA_MAX_SIZE, OFFSETS_RETENTION_CHECK_INTERVAL_MS, OFFSETS_RETENTION_MINUTES);
      }
  
      private ShareGroupConfig createShareGroupConfig() {
          return ShareGroupConfigTest.createShareGroupConfig(SHARE_GROUP_ENABLE, SHARE_GROUP_PARTITION_MAX_RECORD_LOCKS, SHARE_GROUP_DELIVERY_COUNT_LIMIT,
              SHARE_GROUP_RECORD_LOCK_DURATION_MS, SHARE_GROUP_MIN_RECORD_LOCK_DURATION_MS, SHARE_GROUP_MAX_RECORD_LOCK_DURATION_MS);
      }
  }
  