KAFKA-19875 Duplicated topic config prevents broker start

checkstyle/suppressions.xml

@@ -133,6 +133,7 @@
 
     <suppress checks="(WhitespaceAround|LocalVariableName|ImportControl|AvoidStarImport)"
               files="Murmur3Test.java"/>
+    <suppress checks="ImportControl" files="AbstractConfigTest.java"/>
 
     <!-- Connect -->
     <suppress checks="ClassFanOutComplexity"

clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java

@@ -19,6 +19,9 @@
 import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.common.utils.Utils;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -81,7 +84,7 @@
 public class ConfigDef {
 
     private static final Pattern COMMA_WITH_WHITESPACE = Pattern.compile("\\s*,\\s*");
-
+    private static final Logger LOGGER = LoggerFactory.getLogger(ConfigDef.class);
     /**
      * A unique Java object which represents the lack of a default value.
      */
@@ -536,6 +539,14 @@ Object parseValue(ConfigKey key, Object value, boolean isSet) {
             // otherwise assign setting its default value
             parsedValue = key.defaultValue;
         }
+        if (key.validator instanceof ValidList && parsedValue instanceof List) {
+            List<?> originalListValue = (List<?>) parsedValue;
+            parsedValue = originalListValue.stream().distinct().collect(Collectors.toList());
+            if (originalListValue.size() != ((List<?>) parsedValue).size()) {
+                LOGGER.warn("Duplicate configuration \"{}\" values are found. Duplicates will be removed. The original value " +
+                        "is: {}, the updated value is: {}", key.name, originalListValue, parsedValue);
+            }
+        }
         if (key.validator != null) {
             key.validator.ensureValid(key.name, parsedValue);
         }
@@ -1070,8 +1081,10 @@ private void validateIndividualValues(String name, List<Object> values) {
         }
 
         public String toString() {
-            return validString + (isEmptyAllowed ? " (empty config allowed)" : " (empty not allowed)") +
-                    (isNullAllowed ? " (null config allowed)" : " (null not allowed)");
+            String base = validString.validStrings.isEmpty() ? "any non-duplicate values" : validString.toString();
+            String emptyList = isEmptyAllowed ? ", empty list" : "";
+            String nullValue = isNullAllowed ? ", null" : "";
+            return base + emptyList + nullValue;
         }
     }
 

clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java

@@ -16,7 +16,11 @@
  */
 package org.apache.kafka.common.config;
 
+import org.apache.kafka.clients.consumer.ConsumerInterceptor;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.config.provider.EnvVarConfigProvider;
@@ -42,6 +46,7 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -292,22 +297,23 @@ public void testConfiguredInstancesClosedOnFailure() {
 
         try {
             Map<String, String> props = new HashMap<>();
-            String threeConsumerInterceptors = MockConsumerInterceptor.class.getName() + ", "
-                    + MockConsumerInterceptor.class.getName() + ", "
+            String threeConsumerInterceptors = CloseInterceptor.class.getName() + ", "
                     + MockConsumerInterceptor.class.getName();
             props.put(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, threeConsumerInterceptors);
             props.put("client.id", "test");
             TestConfig testConfig = new TestConfig(props);
 
-            MockConsumerInterceptor.setThrowOnConfigExceptionThreshold(3);
+            MockConsumerInterceptor.setThrowOnConfigExceptionThreshold(1);
             assertThrows(
                     Exception.class,
                     () -> testConfig.getConfiguredInstances(TestConfig.METRIC_REPORTER_CLASSES_CONFIG, Object.class)
             );
-            assertEquals(3, MockConsumerInterceptor.CONFIG_COUNT.get());
-            assertEquals(3, MockConsumerInterceptor.CLOSE_COUNT.get());
+            assertEquals(1, MockConsumerInterceptor.CONFIG_COUNT.get());
+            assertEquals(1, MockConsumerInterceptor.CLOSE_COUNT.get());
+            assertEquals(1, CloseInterceptor.CLOSE_COUNT.get());
         } finally {
             MockConsumerInterceptor.resetCounters();
+            CloseInterceptor.resetCounters();
         }
     }
 
@@ -746,4 +752,33 @@ public void configure(Map<String, ?> configs) {
             configs.get(EXTRA_CONFIG);
         }
     }
+
+    public static class CloseInterceptor implements ConsumerInterceptor<String, String> {
+
+        public static final AtomicInteger CLOSE_COUNT = new AtomicInteger(0);
+
+        @Override
+        public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
+            return null;
+        }
+
+        @Override
+        public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
+            // no-op
+        }
+
+        @Override
+        public void close() {
+            CLOSE_COUNT.incrementAndGet();
+        }
+
+        @Override
+        public void configure(Map<String, ?> configs) {
+            // no-op
+        }
+
+        public static void resetCounters() {
+            CLOSE_COUNT.set(0);
+        }
+    }
 }

clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java

@@ -760,34 +760,55 @@ public void testListSizeValidatorToString() {
 
     @Test
     public void testListValidatorAnyNonDuplicateValues() {
-        ConfigDef.ValidList allowAnyNonDuplicateValues = ConfigDef.ValidList.anyNonDuplicateValues(true, true);
-        assertDoesNotThrow(() -> allowAnyNonDuplicateValues.ensureValid("test.config", List.of("a", "b", "c")));
-        assertDoesNotThrow(() -> allowAnyNonDuplicateValues.ensureValid("test.config", List.of()));
-        assertDoesNotThrow(() -> allowAnyNonDuplicateValues.ensureValid("test.config", null));
-        ConfigException exception1 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValues.ensureValid("test.config", List.of("a", "a")));
+        ConfigDef.ValidList allowAnyNonDuplicateValuesAndEmptyListAndNull = ConfigDef.ValidList.anyNonDuplicateValues(true, true);
+        assertDoesNotThrow(() -> allowAnyNonDuplicateValuesAndEmptyListAndNull.ensureValid("test.config", List.of("a", "b", "c")));
+        assertDoesNotThrow(() -> allowAnyNonDuplicateValuesAndEmptyListAndNull.ensureValid("test.config", List.of()));
+        assertDoesNotThrow(() -> allowAnyNonDuplicateValuesAndEmptyListAndNull.ensureValid("test.config", null));
+        ConfigException exception1 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyListAndNull.ensureValid("test.config", List.of("a", "a")));
         assertEquals("Configuration 'test.config' values must not be duplicated.", exception1.getMessage());
-        ConfigException exception2 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValues.ensureValid("test.config", List.of("")));
+        ConfigException exception2 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyListAndNull.ensureValid("test.config", List.of("")));
         assertEquals("Configuration 'test.config' values must not be empty.", exception2.getMessage());
-
+        ConfigException exception3 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyListAndNull.ensureValid("test.config", List.of("a", "", "b")));
+        assertEquals("Configuration 'test.config' values must not be empty.", exception3.getMessage());
+
         ConfigDef.ValidList allowAnyNonDuplicateValuesAndNull = ConfigDef.ValidList.anyNonDuplicateValues(false, true);
         assertDoesNotThrow(() -> allowAnyNonDuplicateValuesAndNull.ensureValid("test.config", List.of("a", "b", "c")));
         assertDoesNotThrow(() -> allowAnyNonDuplicateValuesAndNull.ensureValid("test.config", null));
-        ConfigException exception3 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndNull.ensureValid("test.config", List.of()));
-        assertEquals("Configuration 'test.config' must not be empty. Valid values include: any non-empty value", exception3.getMessage());
-        ConfigException exception4 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndNull.ensureValid("test.config", List.of("a", "a")));
-        assertEquals("Configuration 'test.config' values must not be duplicated.", exception4.getMessage());
-        ConfigException exception5 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndNull.ensureValid("test.config", List.of("")));
-        assertEquals("Configuration 'test.config' values must not be empty.", exception5.getMessage());
+        ConfigException exception4 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndNull.ensureValid("test.config", List.of()));
+        assertEquals("Configuration 'test.config' must not be empty. Valid values include: any non-empty value", exception4.getMessage());
+        ConfigException exception5 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndNull.ensureValid("test.config", List.of("a", "a")));
+        assertEquals("Configuration 'test.config' values must not be duplicated.", exception5.getMessage());
+        ConfigException exception6 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndNull.ensureValid("test.config", List.of("")));
+        assertEquals("Configuration 'test.config' values must not be empty.", exception6.getMessage());
+        ConfigException exception7 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndNull.ensureValid("test.config", List.of("a", "", "b")));
+        assertEquals("Configuration 'test.config' values must not be empty.", exception7.getMessage());
+
 
         ConfigDef.ValidList allowAnyNonDuplicateValuesAndEmptyList = ConfigDef.ValidList.anyNonDuplicateValues(true, false);
         assertDoesNotThrow(() -> allowAnyNonDuplicateValuesAndEmptyList.ensureValid("test.config", List.of("a", "b", "c")));
         assertDoesNotThrow(() -> allowAnyNonDuplicateValuesAndEmptyList.ensureValid("test.config", List.of()));
-        ConfigException exception6 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyList.ensureValid("test.config", null));
-        assertEquals("Configuration 'test.config' values must not be null.", exception6.getMessage());
-        ConfigException exception7 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyList.ensureValid("test.config", List.of("a", "a")));
-        assertEquals("Configuration 'test.config' values must not be duplicated.", exception7.getMessage());
-        ConfigException exception8 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyList.ensureValid("test.config", List.of("")));
-        assertEquals("Configuration 'test.config' values must not be empty.", exception8.getMessage());
+        ConfigException exception8 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyList.ensureValid("test.config", null));
+        assertEquals("Configuration 'test.config' values must not be null.", exception8.getMessage());
+        ConfigException exception9 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyList.ensureValid("test.config", List.of("a", "a")));
+        assertEquals("Configuration 'test.config' values must not be duplicated.", exception9.getMessage());
+        ConfigException exception10 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyList.ensureValid("test.config", List.of("")));
+        assertEquals("Configuration 'test.config' values must not be empty.", exception10.getMessage());
+        ConfigException exception11 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValuesAndEmptyList.ensureValid("test.config", List.of("a", "", "b")));
+        assertEquals("Configuration 'test.config' values must not be empty.", exception11.getMessage());
+
+
+        ConfigDef.ValidList allowAnyNonDuplicateValues = ConfigDef.ValidList.anyNonDuplicateValues(false, false);
+        assertDoesNotThrow(() -> allowAnyNonDuplicateValues.ensureValid("test.config", List.of("a", "b", "c")));
+        ConfigException exception12 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValues.ensureValid("test.config", null));
+        assertEquals("Configuration 'test.config' values must not be null.", exception12.getMessage());
+        ConfigException exception13 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValues.ensureValid("test.config", List.of()));
+        assertEquals("Configuration 'test.config' must not be empty. Valid values include: any non-empty value", exception13.getMessage());
+        ConfigException exception14 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValues.ensureValid("test.config", List.of("a", "", "b")));
+        assertEquals("Configuration 'test.config' values must not be empty.", exception14.getMessage());
+        ConfigException exception15 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValues.ensureValid("test.config", List.of("")));
+        assertEquals("Configuration 'test.config' values must not be empty.", exception15.getMessage());
+        ConfigException exception16 = assertThrows(ConfigException.class, () -> allowAnyNonDuplicateValues.ensureValid("test.config", List.of("a", "", "b")));
+        assertEquals("Configuration 'test.config' values must not be empty.", exception16.getMessage());
     }
 
     @Test

connect/runtime/src/main/java/org/apache/kafka/connect/connector/policy/AllowlistConnectorClientConfigOverridePolicy.java

@@ -40,7 +40,13 @@ public class AllowlistConnectorClientConfigOverridePolicy extends AbstractConnec
     private static final String ALLOWLIST_CONFIG_DOC = "List of client configurations that can be overridden by " +
             "connectors. If empty, connectors can't override any client configurations.";
     private static final ConfigDef CONFIG_DEF = new ConfigDef()
-            .define(ALLOWLIST_CONFIG, ConfigDef.Type.LIST, ALLOWLIST_CONFIG_DEFAULT, ConfigDef.Importance.MEDIUM, ALLOWLIST_CONFIG_DOC);
+            .define(
+                ALLOWLIST_CONFIG,
+                ConfigDef.Type.LIST,
+                ALLOWLIST_CONFIG_DEFAULT,
+                ConfigDef.ValidList.anyNonDuplicateValues(true, false),
+                ConfigDef.Importance.MEDIUM, ALLOWLIST_CONFIG_DOC
+            );
 
     private List<String> allowlist = ALLOWLIST_CONFIG_DEFAULT;
 

core/src/test/scala/unit/kafka/log/LogConfigTest.scala

@@ -294,7 +294,7 @@ class LogConfigTest {
     assertThrows(classOf[ConfigException], () => validateCleanupPolicy())
     logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete")
     assertThrows(classOf[ConfigException], () => validateCleanupPolicy())
-    logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete")
+    logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,delete,delete")
     validateCleanupPolicy()
     logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "")
     validateCleanupPolicy()

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

@@ -220,10 +220,6 @@ class KafkaConfigTest {
     props.setProperty(SocketServerConfigs.LISTENERS_CONFIG, "HOST://localhost:9091,LB://localhost:9092")
     props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, "HOST://localhost:9091,LB://localhost:9091")
     KafkaConfig.fromProps(props)
-
-    // but not duplicate names
-    props.setProperty(SocketServerConfigs.ADVERTISED_LISTENERS_CONFIG, "HOST://localhost:9091,HOST://localhost:9091")
-    assertBadConfigContainingMessage(props, "Configuration 'advertised.listeners' values must not be duplicated.")
   }
 
   @Test
@@ -247,10 +243,6 @@ class KafkaConfigTest {
     caught = assertThrows(classOf[IllegalArgumentException], () => KafkaConfig.fromProps(props))
     assertTrue(caught.getMessage.contains("If you have two listeners on the same port then one needs to be IPv4 and the other IPv6"))
 
-    props.put(SocketServerConfigs.LISTENERS_CONFIG, "PLAINTEXT://127.0.0.1:9092,PLAINTEXT://127.0.0.1:9092")
-    val exception = assertThrows(classOf[ConfigException], () => KafkaConfig.fromProps(props))
-    assertTrue(exception.getMessage.contains("values must not be duplicated."))
-
     props.put(SocketServerConfigs.LISTENERS_CONFIG, "PLAINTEXT://127.0.0.1:9092,SSL://127.0.0.1:9092,SASL_SSL://127.0.0.1:9092")
     caught = assertThrows(classOf[IllegalArgumentException], () => KafkaConfig.fromProps(props))
     assertTrue(caught.getMessage.contains("Each listener must have a different port"))
@@ -301,7 +293,7 @@ class KafkaConfigTest {
     props.setProperty(KRaftConfigs.NODE_ID_CONFIG, "2")
     props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, "2@localhost:9093")
 
-    assertBadConfigContainingMessage(props, 
+    assertBadConfigContainingMessage(props,
       "Missing required configuration \"controller.listener.names\" which has no default value.")
 
     props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "SSL")
@@ -322,7 +314,7 @@ class KafkaConfigTest {
     props.setProperty(QuorumConfig.QUORUM_VOTERS_CONFIG, "2@localhost:9093")
 
     assertFalse(isValidKafkaConfig(props))
-    assertBadConfigContainingMessage(props, 
+    assertBadConfigContainingMessage(props,
       "Missing required configuration \"controller.listener.names\" which has no default value.")
 
     props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "SSL")

docs/upgrade.html

@@ -139,7 +139,9 @@ <h5><a id="upgrade_420_notable" href="#upgrade_420_notable">Notable changes in 4
                 <ul>
                     <li>Null values are no longer accepted for most LIST-type configurations, except those that explicitly
                         allow a null default value or where a null value has a well-defined semantic meaning.</li>
-                    <li>Duplicate entries within the same list are no longer permitted.</li>
+                    <li>Most LIST-type configurations no longer accept duplicate entries, except in cases where duplicates
+                        are explicitly supported. However, if users configure duplicate entries, the internal deduplication
+                        logic will still handle them.</li>
                     <li>Empty lists are no longer allowed, except in configurations where an empty list has a well-defined
                         semantic meaning.</li>
                 </ul>