MINOR: Cleanup Server Common Module (#19085)

Given that now we support Java 17 on our brokers, this PR replace the
use of the following in server-common module:

- Collections.singletonList() and Collections.emptyList() with List.of()
- Collections.singletonMap() and Collections.emptyMap() with Map.of()
- Collections.singleton() and Collections.emptySet() with Set.of()
- Arrays.asList() with List.of()

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

Author: sjhajharia

server-common/src/main/java/org/apache/kafka/server/common/CheckpointFile.java

@@ -31,7 +31,6 @@
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
@@ -146,7 +145,7 @@ public CheckpointReadBuffer(String location,
         public List<T> read() throws IOException {
             String line = reader.readLine();
             if (line == null)
-                return Collections.emptyList();
+                return List.of();
 
             int readVersion = toInt(line);
             if (readVersion != version) {
@@ -156,7 +155,7 @@ public List<T> read() throws IOException {
 
             line = reader.readLine();
             if (line == null) {
-                return Collections.emptyList();
+                return List.of();
             }
             int expectedSize = toInt(line);
             List<T> entries = new ArrayList<>(expectedSize);

server-common/src/main/java/org/apache/kafka/server/common/EligibleLeaderReplicasVersion.java

@@ -16,13 +16,12 @@
  */
 package org.apache.kafka.server.common;
 
-import java.util.Collections;
 import java.util.Map;
 
 public enum EligibleLeaderReplicasVersion implements FeatureVersion {
 
     // Version 0 is the version disable ELR.
-    ELRV_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
+    ELRV_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
 
     // Version 1 enables the ELR (KIP-966).
     ELRV_1(1, MetadataVersion.IBP_4_1_IV0, Map.of(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_4_0_IV1.featureLevel()));

server-common/src/main/java/org/apache/kafka/server/common/FinalizedFeatures.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.server.common;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -27,7 +26,7 @@ public record FinalizedFeatures(
     long finalizedFeaturesEpoch
 ) {
     public static FinalizedFeatures fromKRaftVersion(MetadataVersion version) {
-        return new FinalizedFeatures(version, Collections.emptyMap(), -1);
+        return new FinalizedFeatures(version, Map.of(), -1);
     }
 
     public FinalizedFeatures(

server-common/src/main/java/org/apache/kafka/server/common/GroupVersion.java

@@ -16,16 +16,15 @@
  */
 package org.apache.kafka.server.common;
 
-import java.util.Collections;
 import java.util.Map;
 
 public enum GroupVersion implements FeatureVersion {
 
     // Version 0 is the original group coordinator prior to KIP-848.
-    GV_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
+    GV_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
 
     // Version 1 enables the consumer rebalance protocol (KIP-848).
-    GV_1(1, MetadataVersion.IBP_4_0_IV0, Collections.emptyMap());
+    GV_1(1, MetadataVersion.IBP_4_0_IV0, Map.of());
 
     public static final String FEATURE_NAME = "group.version";
 

server-common/src/main/java/org/apache/kafka/server/common/KRaftVersion.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.server.common;
 
-import java.util.Collections;
 import java.util.Map;
 
 public enum KRaftVersion implements FeatureVersion {
@@ -74,9 +73,9 @@ public MetadataVersion bootstrapMetadataVersion() {
     @Override
     public Map<String, Short> dependencies() {
         if (this.featureLevel == 0) {
-            return Collections.emptyMap();
+            return Map.of();
         } else {
-            return Collections.singletonMap(
+            return Map.of(
                 MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_9_IV0.featureLevel());
         }
     }

server-common/src/main/java/org/apache/kafka/server/common/TestFeatureVersion.java

@@ -16,15 +16,14 @@
  */
 package org.apache.kafka.server.common;
 
-import java.util.Collections;
 import java.util.Map;
 
 public enum TestFeatureVersion implements FeatureVersion {
-    TEST_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
+    TEST_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
     // TEST_1 released right before MV 3.7-IVO was released, and it has no dependencies
-    TEST_1(1, MetadataVersion.IBP_3_7_IV0, Collections.emptyMap()),
+    TEST_1(1, MetadataVersion.IBP_3_7_IV0, Map.of()),
     // TEST_2 is not yet set to be the default version and maps to the latest testing version, and it depends on this metadata version
-    TEST_2(2, MetadataVersion.latestTesting(), Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.latestTesting().featureLevel()));
+    TEST_2(2, MetadataVersion.latestTesting(), Map.of(MetadataVersion.FEATURE_NAME, MetadataVersion.latestTesting().featureLevel()));
 
     private final short featureLevel;
     private final MetadataVersion metadataVersionMapping;

server-common/src/main/java/org/apache/kafka/server/common/TransactionVersion.java

@@ -19,17 +19,16 @@
 import org.apache.kafka.common.requests.AddPartitionsToTxnRequest;
 import org.apache.kafka.common.requests.EndTxnRequest;
 
-import java.util.Collections;
 import java.util.Map;
 
 public enum TransactionVersion implements FeatureVersion {
 
     // Version 0 is the original transaction coordinator with no extra features enabled.
-    TV_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
+    TV_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
     // Version 1 enables flexible transactional state records. (KIP-890)
-    TV_1(1, MetadataVersion.IBP_4_0_IV2, Collections.emptyMap()),
+    TV_1(1, MetadataVersion.IBP_4_0_IV2, Map.of()),
     // Version 2 enables epoch bump per transaction and optimizations. (KIP-890)
-    TV_2(2, MetadataVersion.IBP_4_0_IV2, Collections.emptyMap());
+    TV_2(2, MetadataVersion.IBP_4_0_IV2, Map.of());
 
     public static final String FEATURE_NAME = "transaction.version";
 

server-common/src/main/java/org/apache/kafka/server/common/UnitTestFeatureVersion.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.server.common;
 
-import java.util.Collections;
 import java.util.Map;
 
 /**
@@ -27,8 +26,8 @@ public class UnitTestFeatureVersion {
      * The feature is used for testing latest production is not one of the feature versions.
      */
     public enum FV0 implements FeatureVersion {
-        UT_FV0_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
-        UT_FV0_1(1, MetadataVersion.IBP_3_7_IV0, Collections.emptyMap());
+        UT_FV0_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
+        UT_FV0_1(1, MetadataVersion.IBP_3_7_IV0, Map.of());
 
         private final short featureLevel;
         private final MetadataVersion bootstrapMetadataVersion;
@@ -69,8 +68,8 @@ public Map<String, Short> dependencies() {
      * The feature is used to test latest production lags behind the default value.
      */
     public enum FV1 implements FeatureVersion {
-        UT_FV1_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
-        UT_FV1_1(1, MetadataVersion.IBP_3_7_IV0, Collections.emptyMap());
+        UT_FV1_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
+        UT_FV1_1(1, MetadataVersion.IBP_3_7_IV0, Map.of());
 
         private final short featureLevel;
         private final MetadataVersion bootstrapMetadataVersion;
@@ -111,8 +110,8 @@ public Map<String, Short> dependencies() {
      * The feature is used to test the dependency of the latest production that is not yet production ready.
      */
     public enum FV2 implements FeatureVersion {
-        UT_FV2_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
-        UT_FV2_1(1, MetadataVersion.IBP_3_7_IV0, Collections.emptyMap());
+        UT_FV2_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
+        UT_FV2_1(1, MetadataVersion.IBP_3_7_IV0, Map.of());
 
         private final short featureLevel;
         private final MetadataVersion bootstrapMetadataVersion;
@@ -153,8 +152,8 @@ public Map<String, Short> dependencies() {
      * The feature is used to test the dependency of the latest production that is not yet production ready.
      */
     public enum FV3 implements FeatureVersion {
-        UT_FV3_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
-        UT_FV3_1(1, MetadataVersion.IBP_3_7_IV0, Collections.singletonMap(FV2.FEATURE_NAME, (short) 1));
+        UT_FV3_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
+        UT_FV3_1(1, MetadataVersion.IBP_3_7_IV0, Map.of(FV2.FEATURE_NAME, (short) 1));
 
         private final short featureLevel;
         private final MetadataVersion bootstrapMetadataVersion;
@@ -195,8 +194,8 @@ public Map<String, Short> dependencies() {
      * The feature is used to test the dependency of the default value that is not yet default ready.
      */
     public enum FV4 implements FeatureVersion {
-        UT_FV4_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
-        UT_FV4_1(1, MetadataVersion.latestTesting(), Collections.emptyMap());
+        UT_FV4_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
+        UT_FV4_1(1, MetadataVersion.latestTesting(), Map.of());
 
         private final short featureLevel;
         private final MetadataVersion bootstrapMetadataVersion;
@@ -237,8 +236,8 @@ public Map<String, Short> dependencies() {
      * The feature is used to test the dependency of the default value that is not yet default ready.
      */
     public enum FV5 implements FeatureVersion {
-        UT_FV5_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
-        UT_FV5_1(1, MetadataVersion.IBP_3_7_IV0, Collections.singletonMap(FV4.FEATURE_NAME, (short) 1));
+        UT_FV5_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
+        UT_FV5_1(1, MetadataVersion.IBP_3_7_IV0, Map.of(FV4.FEATURE_NAME, (short) 1));
 
         private final short featureLevel;
         private final MetadataVersion bootstrapMetadataVersion;
@@ -279,8 +278,8 @@ public Map<String, Short> dependencies() {
      * The feature is used to test the latest production has MV dependency that is not yet production ready.
      */
     public enum FV6 implements FeatureVersion {
-        UT_FV6_0(0, MetadataVersion.MINIMUM_VERSION, Collections.emptyMap()),
-        UT_FV6_1(1, MetadataVersion.latestTesting(), Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.latestTesting().featureLevel()));
+        UT_FV6_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
+        UT_FV6_1(1, MetadataVersion.latestTesting(), Map.of(MetadataVersion.FEATURE_NAME, MetadataVersion.latestTesting().featureLevel()));
 
         private final short featureLevel;
         private final MetadataVersion bootstrapMetadataVersion;
@@ -321,8 +320,8 @@ public Map<String, Short> dependencies() {
      * The feature is used to test the default value has MV dependency that is behind the bootstrap MV.
      */
     public enum FV7 implements FeatureVersion {
-        UT_FV7_0(0, MetadataVersion.MINIMUM_VERSION, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
-        UT_FV7_1(1, MetadataVersion.IBP_3_8_IV0, Collections.singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_8_IV0.featureLevel()));
+        UT_FV7_0(0, MetadataVersion.MINIMUM_VERSION, Map.of(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_7_IV0.featureLevel())),
+        UT_FV7_1(1, MetadataVersion.IBP_3_8_IV0, Map.of(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_8_IV0.featureLevel()));
 
         private final short featureLevel;
         private final MetadataVersion bootstrapMetadataVersion;

server-common/src/main/java/org/apache/kafka/server/config/QuotaConfig.java

@@ -20,7 +20,6 @@
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.security.scram.internals.ScramMechanism;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 
@@ -62,14 +61,14 @@ public class QuotaConfig {
             "the leader side. The list should describe a set of replicas in the form " +
             "[PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... or alternatively the wildcard '*' can be used to throttle " +
             "all replicas for this topic.";
-    public static final List<String> LEADER_REPLICATION_THROTTLED_REPLICAS_DEFAULT = Collections.emptyList();
+    public static final List<String> LEADER_REPLICATION_THROTTLED_REPLICAS_DEFAULT = List.of();
 
     public static final String FOLLOWER_REPLICATION_THROTTLED_REPLICAS_CONFIG = "follower.replication.throttled.replicas";
     public static final String FOLLOWER_REPLICATION_THROTTLED_REPLICAS_DOC = "A list of replicas for which log replication should be throttled on " +
             "the follower side. The list should describe a set of " + "replicas in the form " +
             "[PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... or alternatively the wildcard '*' can be used to throttle " +
             "all replicas for this topic.";
-    public static final List<String> FOLLOWER_REPLICATION_THROTTLED_REPLICAS_DEFAULT = Collections.emptyList();
+    public static final List<String> FOLLOWER_REPLICATION_THROTTLED_REPLICAS_DEFAULT = List.of();
 
 
     public static final String LEADER_REPLICATION_THROTTLED_RATE_CONFIG = "leader.replication.throttled.rate";

server-common/src/main/java/org/apache/kafka/server/config/ServerTopicConfigSynonyms.java

@@ -20,14 +20,12 @@
 import org.apache.kafka.common.utils.Utils;
 
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
 import java.util.stream.Collectors;
 
-import static java.util.Arrays.asList;
 
 public final class ServerTopicConfigSynonyms {
     public static final String LOG_PREFIX = "log.";
@@ -50,7 +48,7 @@ public final class ServerTopicConfigSynonyms {
      * the first synonym and ignore the second.
      */
     // Topic configs with no mapping to a server config can be found in `LogConfig.CONFIGS_WITH_NO_SERVER_DEFAULTS`
-    public static final Map<String, List<ConfigSynonym>> ALL_TOPIC_CONFIG_SYNONYMS = Collections.unmodifiableMap(Utils.mkMap(
+    public static final Map<String, List<ConfigSynonym>> ALL_TOPIC_CONFIG_SYNONYMS = Utils.mkMap(
         sameNameWithLogPrefix(TopicConfig.SEGMENT_BYTES_CONFIG),
         listWithLogPrefix(TopicConfig.SEGMENT_MS_CONFIG,
             new ConfigSynonym("roll.ms"),
@@ -88,17 +86,17 @@ public final class ServerTopicConfigSynonyms {
         sameNameWithLogPrefix(TopicConfig.MESSAGE_TIMESTAMP_AFTER_MAX_MS_CONFIG),
         sameNameWithLogPrefix(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG),
         sameNameWithLogPrefix(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG)
-    ));
+    );
 
     /**
      * Map topic config to the server config with the highest priority. Some of these have additional
      * synonyms that can be obtained using [[kafka.server.DynamicBrokerConfig#brokerConfigSynonyms]]
      * or using [[AllTopicConfigSynonyms]]
      */
-    public static final Map<String, String> TOPIC_CONFIG_SYNONYMS = Collections.unmodifiableMap(
+    public static final Map<String, String> TOPIC_CONFIG_SYNONYMS =
         ALL_TOPIC_CONFIG_SYNONYMS.entrySet()
         .stream()
-        .collect(Collectors.toMap(Entry::getKey, e -> e.getValue().get(0).name())));
+        .collect(Collectors.toMap(Entry::getKey, e -> e.getValue().get(0).name()));
 
     /**
      * Return the server config with the highest priority for `topicConfigName` if it exists. Otherwise,
@@ -112,23 +110,23 @@ public static String serverSynonym(String topicConfigName) {
     }
 
     private static Entry<String, List<ConfigSynonym>> sameName(String configName) {
-        return Utils.mkEntry(configName, asList(new ConfigSynonym(configName)));
+        return Utils.mkEntry(configName, List.of(new ConfigSynonym(configName)));
     }
 
     private static Entry<String, List<ConfigSynonym>> sameNameWithLogPrefix(String configName) {
-        return Utils.mkEntry(configName, asList(new ConfigSynonym(LOG_PREFIX + configName)));
+        return Utils.mkEntry(configName, List.of(new ConfigSynonym(LOG_PREFIX + configName)));
     }
 
     private static Entry<String, List<ConfigSynonym>> sameNameWithLogCleanerPrefix(String configName) {
-        return Utils.mkEntry(configName, asList(new ConfigSynonym(LOG_CLEANER_PREFIX + configName)));
+        return Utils.mkEntry(configName, List.of(new ConfigSynonym(LOG_CLEANER_PREFIX + configName)));
     }
 
     private static Entry<String, List<ConfigSynonym>> singleWithLogPrefix(String topicConfigName, String brokerConfigName) {
-        return Utils.mkEntry(topicConfigName, asList(new ConfigSynonym(LOG_PREFIX + brokerConfigName)));
+        return Utils.mkEntry(topicConfigName, List.of(new ConfigSynonym(LOG_PREFIX + brokerConfigName)));
     }
 
     private static Entry<String, List<ConfigSynonym>> singleWithLogCleanerPrefix(String topicConfigName, String brokerConfigName) {
-        return Utils.mkEntry(topicConfigName, asList(new ConfigSynonym(LOG_CLEANER_PREFIX + brokerConfigName)));
+        return Utils.mkEntry(topicConfigName, List.of(new ConfigSynonym(LOG_CLEANER_PREFIX + brokerConfigName)));
     }
 
     private static Entry<String, List<ConfigSynonym>> listWithLogPrefix(String topicConfigName, ConfigSynonym... synonyms) {
@@ -139,6 +137,6 @@ private static Entry<String, List<ConfigSynonym>> listWithLogPrefix(String topic
     }
 
     private static Entry<String, List<ConfigSynonym>> single(String topicConfigName, String brokerConfigName) {
-        return Utils.mkEntry(topicConfigName, asList(new ConfigSynonym(brokerConfigName)));
+        return Utils.mkEntry(topicConfigName, List.of(new ConfigSynonym(brokerConfigName)));
     }
 }