MINOR: Move populateSynonyms from KafkaConfig to AbstractKafkaConfig (#21381)

Move `populateSynonyms` from `KafkaConfig` to `AbstractKafkaConfig` and
migrate the related tests.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

Author: majialoong

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -128,21 +128,6 @@ object KafkaConfig {
     }
     if (maybeSensitive) Password.HIDDEN else value
   }
-
-  /**
-   * Copy a configuration map, populating some keys that we want to treat as synonyms.
-   */
-  def populateSynonyms(input: util.Map[_, _]): util.Map[Any, Any] = {
-    val output = new util.HashMap[Any, Any](input)
-    val brokerId = output.get(ServerConfigs.BROKER_ID_CONFIG)
-    val nodeId = output.get(KRaftConfigs.NODE_ID_CONFIG)
-    if (brokerId == null && nodeId != null) {
-      output.put(ServerConfigs.BROKER_ID_CONFIG, nodeId)
-    } else if (brokerId != null && nodeId == null) {
-      output.put(KRaftConfigs.NODE_ID_CONFIG, brokerId)
-    }
-    output
-  }
 }
 
 /**
@@ -154,8 +139,8 @@ object KafkaConfig {
 class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
   extends AbstractKafkaConfig(KafkaConfig.configDef, props, Utils.castToStringObjectMap(props), doLog) with Logging {
 
-  def this(props: java.util.Map[_, _]) = this(true, KafkaConfig.populateSynonyms(props))
-  def this(props: java.util.Map[_, _], doLog: Boolean) = this(doLog, KafkaConfig.populateSynonyms(props))
+  def this(props: java.util.Map[_, _]) = this(true, AbstractKafkaConfig.populateSynonyms(props))
+  def this(props: java.util.Map[_, _], doLog: Boolean) = this(doLog, AbstractKafkaConfig.populateSynonyms(props))
 
   // Cache the current config to avoid acquiring read lock to access from dynamicConfig
   @volatile private var currentConfig = this

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

@@ -1574,31 +1574,6 @@ class KafkaConfigTest {
     assertEquals(util.List.of(dataDir1, dataDir2), config.logDirs)
   }
 
-  @Test
-  def testPopulateSynonymsOnEmptyMap(): Unit = {
-    assertEquals(Collections.emptyMap(), KafkaConfig.populateSynonyms(Collections.emptyMap()))
-  }
-
-  @Test
-  def testPopulateSynonymsOnMapWithoutNodeId(): Unit = {
-    val input =  new util.HashMap[String, String]()
-    input.put(ServerConfigs.BROKER_ID_CONFIG, "4")
-    val expectedOutput = new util.HashMap[String, String]()
-    expectedOutput.put(ServerConfigs.BROKER_ID_CONFIG, "4")
-    expectedOutput.put(KRaftConfigs.NODE_ID_CONFIG, "4")
-    assertEquals(expectedOutput, KafkaConfig.populateSynonyms(input))
-  }
-
-  @Test
-  def testPopulateSynonymsOnMapWithoutBrokerId(): Unit = {
-    val input =  new util.HashMap[String, String]()
-    input.put(KRaftConfigs.NODE_ID_CONFIG, "4")
-    val expectedOutput = new util.HashMap[String, String]()
-    expectedOutput.put(ServerConfigs.BROKER_ID_CONFIG, "4")
-    expectedOutput.put(KRaftConfigs.NODE_ID_CONFIG, "4")
-    assertEquals(expectedOutput, KafkaConfig.populateSynonyms(input))
-  }
-
   @Test
   def testNodeIdMustNotBeDifferentThanBrokerId(): Unit = {
     val props = new Properties()

server/src/main/java/org/apache/kafka/server/config/AbstractKafkaConfig.java

@@ -44,6 +44,7 @@
 
 import org.apache.commons.validator.routines.InetAddressValidator;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -186,6 +187,21 @@ public static Map<String, String> getMap(String propName, String propValue) {
         }
     }
 
+    /**
+     * Copy a configuration map, populating some keys that we want to treat as synonyms.
+     */
+    public static Map<Object, Object> populateSynonyms(Map<?, ?> input) {
+        Map<Object, Object> output = new HashMap<>(input);
+        Object brokerId = output.get(ServerConfigs.BROKER_ID_CONFIG);
+        Object nodeId = output.get(KRaftConfigs.NODE_ID_CONFIG);
+        if (brokerId == null && nodeId != null) {
+            output.put(ServerConfigs.BROKER_ID_CONFIG, nodeId);
+        } else if (brokerId != null && nodeId == null) {
+            output.put(KRaftConfigs.NODE_ID_CONFIG, brokerId);
+        }
+        return output;
+    }
+
     public static List<Endpoint> listenerListToEndPoints(List<String> listeners, Map<ListenerName, SecurityProtocol> securityProtocolMap) {
         return listenerListToEndPoints(listeners, securityProtocolMap, true);
     }

server/src/test/java/org/apache/kafka/server/config/AbstractKafkaConfigTest.java

@@ -0,0 +1,55 @@
+/*
+ * 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.server.config;
+
+import org.apache.kafka.raft.KRaftConfigs;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class AbstractKafkaConfigTest {
+
+    @Test
+    public void testPopulateSynonymsOnEmptyMap() {
+        assertEquals(Collections.emptyMap(), AbstractKafkaConfig.populateSynonyms(Collections.emptyMap()));
+    }
+
+    @Test
+    public void testPopulateSynonymsOnMapWithoutNodeId() {
+        Map<String, String> input = new HashMap<>();
+        input.put(ServerConfigs.BROKER_ID_CONFIG, "4");
+        Map<String, String> expectedOutput = new HashMap<>();
+        expectedOutput.put(ServerConfigs.BROKER_ID_CONFIG, "4");
+        expectedOutput.put(KRaftConfigs.NODE_ID_CONFIG, "4");
+        assertEquals(expectedOutput, AbstractKafkaConfig.populateSynonyms(input));
+    }
+
+    @Test
+    public void testPopulateSynonymsOnMapWithoutBrokerId() {
+        Map<String, String> input = new HashMap<>();
+        input.put(KRaftConfigs.NODE_ID_CONFIG, "4");
+        Map<String, String> expectedOutput = new HashMap<>();
+        expectedOutput.put(ServerConfigs.BROKER_ID_CONFIG, "4");
+        expectedOutput.put(KRaftConfigs.NODE_ID_CONFIG, "4");
+        assertEquals(expectedOutput, AbstractKafkaConfig.populateSynonyms(input));
+    }
+}