MINOR: Cleanup Storage Module (#19072)

Given that now we support Java 17 on our brokers, this PR replace the
use of the following in storage 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: Ken Huang <[email protected]>, Chia-Ping Tsai
<[email protected]>

Author: sjhajharia

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java

@@ -85,11 +85,11 @@ class ConsumerTask implements Runnable, Closeable {
     private final Object assignPartitionsLock = new Object();
 
     // Remote log metadata topic partitions that consumer is assigned to.
-    private volatile Set<Integer> assignedMetadataPartitions = Collections.emptySet();
+    private volatile Set<Integer> assignedMetadataPartitions = Set.of();
 
     // User topic partitions that this broker is a leader/follower for.
-    private volatile Map<TopicIdPartition, UserTopicIdPartition> assignedUserTopicIdPartitions = Collections.emptyMap();
-    private volatile Set<TopicIdPartition> processedAssignmentOfUserTopicIdPartitions = Collections.emptySet();
+    private volatile Map<TopicIdPartition, UserTopicIdPartition> assignedUserTopicIdPartitions = Map.of();
+    private volatile Set<TopicIdPartition> processedAssignmentOfUserTopicIdPartitions = Set.of();
 
     private long uninitializedAt;
     private boolean isAllUserTopicPartitionsInitialized;
@@ -299,11 +299,11 @@ private void clearResourcesForUnassignedUserTopicPartitions(Set<TopicIdPartition
     }
 
     void addAssignmentsForPartitions(final Set<TopicIdPartition> partitions) {
-        updateAssignments(Objects.requireNonNull(partitions), Collections.emptySet());
+        updateAssignments(Objects.requireNonNull(partitions), Set.of());
     }
 
     void removeAssignmentsForPartitions(final Set<TopicIdPartition> partitions) {
-        updateAssignments(Collections.emptySet(), Objects.requireNonNull(partitions));
+        updateAssignments(Set.of(), Objects.requireNonNull(partitions));
     }
 
     private void updateAssignments(final Set<TopicIdPartition> addedPartitions,

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataStore.java

@@ -31,7 +31,6 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.Collections;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
@@ -154,8 +153,8 @@ public void close() throws IOException {
 
         // Clear the entries by creating unmodifiable empty maps.
         // Practically, we do not use the same instances that are closed.
-        idToPartitionDeleteMetadata = Collections.emptyMap();
-        idToRemoteLogMetadataCache = Collections.emptyMap();
+        idToPartitionDeleteMetadata = Map.of();
+        idToRemoteLogMetadataCache = Map.of();
     }
 
     @Override

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManager.java

@@ -58,7 +58,6 @@
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Function;
 import java.util.function.Supplier;
-import java.util.stream.Collectors;
 
 /**
  * This is the {@link RemoteLogMetadataManager} implementation with storage as an internal topic with name {@link TopicBasedRemoteLogMetadataManagerConfig#REMOTE_LOG_METADATA_TOPIC_NAME}.
@@ -471,7 +470,7 @@ private void initializeResources() {
 
     boolean doesTopicExist(Admin adminClient, String topic) {
         try {
-            TopicDescription description = adminClient.describeTopics(Collections.singleton(topic))
+            TopicDescription description = adminClient.describeTopics(Set.of(topic))
                     .topicNameValues()
                     .get(topic)
                     .get();
@@ -491,7 +490,7 @@ boolean doesTopicExist(Admin adminClient, String topic) {
     private boolean isPartitionsCountSameAsConfigured(Admin adminClient,
                                                       String topicName) throws InterruptedException, ExecutionException {
         log.debug("Getting topic details to check for partition count and replication factor.");
-        TopicDescription topicDescription = adminClient.describeTopics(Collections.singleton(topicName))
+        TopicDescription topicDescription = adminClient.describeTopics(Set.of(topicName))
                                                        .topicNameValues().get(topicName).get();
         int expectedPartitions = rlmmConfig.metadataTopicPartitionsCount();
         int topicPartitionsSize = topicDescription.partitions().size();
@@ -525,14 +524,14 @@ private boolean createTopic(Admin adminClient, NewTopic newTopic) {
         try {
             doesTopicExist = doesTopicExist(adminClient, topic);
             if (!doesTopicExist) {
-                CreateTopicsResult result = adminClient.createTopics(Collections.singleton(newTopic));
+                CreateTopicsResult result = adminClient.createTopics(Set.of(newTopic));
                 result.all().get();
                 List<String> overriddenConfigs = result.config(topic).get()
                         .entries()
                         .stream()
                         .filter(entry -> !entry.isDefault())
                         .map(entry -> entry.name() + "=" + entry.value())
-                        .collect(Collectors.toList());
+                        .toList();
                 log.info("Topic {} created. TopicId: {}, numPartitions: {}, replicationFactor: {}, config: {}",
                         topic, result.topicId(topic).get(), result.numPartitions(topic).get(),
                         result.replicationFactor(topic).get(), overriddenConfigs);

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/serialization/RemoteLogSegmentMetadataSnapshotTransform.java

@@ -26,7 +26,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.stream.Collectors;
 
 public class RemoteLogSegmentMetadataSnapshotTransform implements RemoteLogMetadataTransform<RemoteLogSegmentMetadataSnapshot> {
 
@@ -52,7 +51,7 @@ private List<RemoteLogSegmentMetadataSnapshotRecord.SegmentLeaderEpochEntry> cre
                            .map(entry -> new RemoteLogSegmentMetadataSnapshotRecord.SegmentLeaderEpochEntry()
                            .setLeaderEpoch(entry.getKey())
                            .setOffset(entry.getValue()))
-                           .collect(Collectors.toList());
+                           .toList();
     }
 
     @Override

storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/serialization/RemoteLogSegmentMetadataTransform.java

@@ -30,7 +30,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.stream.Collectors;
 
 public class RemoteLogSegmentMetadataTransform implements RemoteLogMetadataTransform<RemoteLogSegmentMetadata> {
 
@@ -56,7 +55,7 @@ private List<RemoteLogSegmentMetadataRecord.SegmentLeaderEpochEntry> createSegme
                    .map(entry -> new RemoteLogSegmentMetadataRecord.SegmentLeaderEpochEntry()
                            .setLeaderEpoch(entry.getKey())
                            .setOffset(entry.getValue()))
-                   .collect(Collectors.toList());
+                   .toList();
     }
 
     private RemoteLogSegmentMetadataRecord.RemoteLogSegmentIdEntry createRemoteLogSegmentIdEntry(RemoteLogSegmentMetadata data) {

storage/src/main/java/org/apache/kafka/server/log/remote/quota/RLMQuotaManager.java

@@ -32,7 +32,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -102,7 +101,7 @@ private MetricConfig getQuotaMetricConfig(Quota quota) {
     }
 
     private MetricName metricName() {
-        return metrics.metricName("byte-rate", quotaType.toString(), description, Collections.emptyMap());
+        return metrics.metricName("byte-rate", quotaType.toString(), description, Map.of());
     }
 
     private Sensor sensor() {

storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java

@@ -454,7 +454,7 @@ public Map<String, Object> remoteLogMetadataManagerProps() {
 
     public Map<String, Object> getConfigProps(String configPrefixProp) {
         String prefixProp = config.getString(configPrefixProp);
-        return prefixProp == null ? Collections.emptyMap() : Collections.unmodifiableMap(config.originalsWithPrefix(prefixProp));
+        return prefixProp == null ? Map.of() : Collections.unmodifiableMap(config.originalsWithPrefix(prefixProp));
     }
 
     public int remoteLogManagerCopyNumQuotaSamples() {

storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java

@@ -26,7 +26,6 @@
 
 import java.util.AbstractMap;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -406,7 +405,7 @@ private static List<EpochEntry> truncateFromEnd(TreeMap<Integer, EpochEntry> epo
         if (endOffset >= 0 && epochEntry.isPresent() && epochEntry.get().startOffset >= endOffset) {
             return removeWhileMatching(epochs.descendingMap().entrySet().iterator(), x -> x.startOffset >= endOffset);
         }
-        return Collections.emptyList();
+        return List.of();
     }
 
     public OptionalInt epochForOffset(long offset) {

storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java

@@ -36,7 +36,6 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -51,7 +50,6 @@
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
-import static java.util.Collections.singletonList;
 import static org.apache.kafka.common.utils.Utils.require;
 import static org.apache.kafka.storage.internals.log.LogFileUtils.CLEANED_FILE_SUFFIX;
 import static org.apache.kafka.storage.internals.log.LogFileUtils.DELETED_FILE_SUFFIX;
@@ -433,11 +431,11 @@ public LogSegment createAndDeleteSegment(long newOffset,
                 config.preallocate);
         segments.add(newSegment);
 
-        reason.logReason(singletonList(segmentToDelete));
+        reason.logReason(List.of(segmentToDelete));
         if (newOffset != segmentToDelete.baseOffset()) {
             segments.remove(segmentToDelete.baseOffset());
         }
-        deleteSegmentFiles(singletonList(segmentToDelete), asyncDelete, dir, topicPartition, config, scheduler, logDirFailureChannel, logIdent);
+        deleteSegmentFiles(List.of(segmentToDelete), asyncDelete, dir, topicPartition, config, scheduler, logDirFailureChannel, logIdent);
         return newSegment;
     }
 
@@ -619,7 +617,7 @@ public LogSegment roll(Long expectedNextOffset) {
                     File offsetIdxFile = LogFileUtils.offsetIndexFile(dir, newOffset);
                     File timeIdxFile = LogFileUtils.timeIndexFile(dir, newOffset);
                     File txnIdxFile = LogFileUtils.transactionIndexFile(dir, newOffset);
-                    for (File file : Arrays.asList(logFile, offsetIdxFile, timeIdxFile, txnIdxFile)) {
+                    for (File file : List.of(logFile, offsetIdxFile, timeIdxFile, txnIdxFile)) {
                         if (file.exists()) {
                             logger.warn("Newly rolled segment file {} already exists; deleting it first", file.getAbsolutePath());
                             Files.delete(file.toPath());
@@ -791,7 +789,7 @@ public static <T> Optional<T> nextItem(Iterator<T> iterator) {
 
     private static FetchDataInfo emptyFetchDataInfo(LogOffsetMetadata fetchOffsetMetadata, boolean includeAbortedTxns) {
         Optional<List<FetchResponseData.AbortedTransaction>> abortedTransactions = includeAbortedTxns
-            ? Optional.of(Collections.emptyList())
+            ? Optional.of(List.of())
             : Optional.empty();
         return new FetchDataInfo(fetchOffsetMetadata, MemoryRecords.EMPTY, false, abortedTransactions);
     }
@@ -943,7 +941,7 @@ public static SplitSegmentResult splitOverflowedSegment(LogSegment segment,
             }
             // replace old segment with new ones
             LOG.info("{}Replacing overflowed segment {} with split segments {}", logPrefix, segment, newSegments);
-            List<LogSegment> deletedSegments = replaceSegments(existingSegments, newSegments, singletonList(segment),
+            List<LogSegment> deletedSegments = replaceSegments(existingSegments, newSegments, List.of(segment),
                     dir, topicPartition, config, scheduler, logDirFailureChannel, logPrefix, false);
             return new SplitSegmentResult(deletedSegments, newSegments);
         } catch (Exception e) {
@@ -1035,7 +1033,7 @@ public static List<LogSegment> replaceSegments(LogSegments existingSegments,
                 existingSegments.remove(segment.baseOffset());
             }
             deleteSegmentFiles(
-                    singletonList(segment),
+                    List.of(segment),
                     true,
                     dir,
                     topicPartition,

storage/src/main/java/org/apache/kafka/storage/internals/log/LogAppendInfo.java

@@ -21,7 +21,6 @@
 import org.apache.kafka.common.record.RecordValidationStats;
 import org.apache.kafka.common.requests.ProduceResponse.RecordError;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
@@ -74,7 +73,7 @@ public LogAppendInfo(long firstOffset,
                          int validBytes,
                          long lastOffsetOfFirstBatch) {
         this(firstOffset, lastOffset, lastLeaderEpoch, maxTimestamp, logAppendTime, logStartOffset,
-            recordValidationStats, sourceCompression, validBytes, lastOffsetOfFirstBatch, Collections.emptyList(),
+            recordValidationStats, sourceCompression, validBytes, lastOffsetOfFirstBatch, List.of(),
                 LeaderHwChange.NONE);
     }
 

storage/src/main/java/org/apache/kafka/storage/internals/log/LogConfig.java

@@ -48,7 +48,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
-import static java.util.Arrays.asList;
 import static org.apache.kafka.common.config.ConfigDef.Importance.HIGH;
 import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
 import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
@@ -105,7 +104,7 @@ public LogConfigDef(ConfigDef base) {
 
         @Override
         public List<String> headers() {
-            return asList("Name", "Description", "Type", "Default", "Valid Values", SERVER_DEFAULT_HEADER_NAME, "Importance");
+            return List.of("Name", "Description", "Type", "Default", "Valid Values", SERVER_DEFAULT_HEADER_NAME, "Importance");
         }
 
         // Visible for testing
@@ -300,7 +299,7 @@ public Optional<String> serverConfigName(String configName) {
     private final Map<?, ?> props;
 
     public LogConfig(Map<?, ?> props) {
-        this(props, Collections.emptySet());
+        this(props, Set.of());
     }
 
     @SuppressWarnings({"this-escape"})
@@ -326,11 +325,11 @@ public LogConfig(Map<?, ?> props, Set<String> overriddenConfigs) {
         this.minCleanableRatio = getDouble(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG);
         this.compact = getList(TopicConfig.CLEANUP_POLICY_CONFIG).stream()
                 .map(c -> c.toLowerCase(Locale.ROOT))
-                .collect(Collectors.toList())
+                .toList()
                 .contains(TopicConfig.CLEANUP_POLICY_COMPACT);
         this.delete = getList(TopicConfig.CLEANUP_POLICY_CONFIG).stream()
                 .map(c -> c.toLowerCase(Locale.ROOT))
-                .collect(Collectors.toList())
+                .toList()
                 .contains(TopicConfig.CLEANUP_POLICY_DELETE);
         this.uncleanLeaderElectionEnable = getBoolean(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG);
         this.minInSyncReplicas = getInt(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG);
@@ -447,7 +446,7 @@ public static Optional<Type> configType(String configName) {
     }
 
     public static List<String> configNames() {
-        return CONFIG.names().stream().sorted().collect(Collectors.toList());
+        return CONFIG.names().stream().sorted().toList();
     }
 
     public static Optional<String> serverConfigName(String configName) {
@@ -610,7 +609,7 @@ private static void validateRemoteStorageRetentionTime(Map<?, ?> props) {
      * Check that the given properties contain only valid log config names and that all values can be parsed and are valid
      */
     public static void validate(Properties props) {
-        validate(Collections.emptyMap(), props, Collections.emptyMap(), false);
+        validate(Map.of(), props, Map.of(), false);
     }
 
     public static void validate(Map<String, String> existingConfigs,

storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java

@@ -43,14 +43,14 @@
 import java.nio.file.NoSuchFileException;
 import java.nio.file.attribute.FileTime;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Optional;
 import java.util.OptionalLong;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-import static java.util.Arrays.asList;
 
 /**
  * A segment of the log. Each segment has two components: a log and an index. The log is a FileRecords containing
@@ -772,7 +772,7 @@ void closeHandlers() {
      */
     public void deleteIfExists() throws IOException {
         try {
-            Utils.tryAll(asList(
+            Utils.tryAll(List.of(
                 () -> deleteTypeIfExists(() -> log.deleteIfExists(), "log", log.file(), true),
                 () -> deleteTypeIfExists(() -> lazyOffsetIndex.deleteIfExists(), "offset index", offsetIndexFile(), true),
                 () -> deleteTypeIfExists(() -> lazyTimeIndex.deleteIfExists(), "time index", timeIndexFile(), true),

storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegments.java

@@ -22,14 +22,13 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.Collection;
-import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalLong;
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.function.Predicate;
-import java.util.stream.Collectors;
 
 /**
  * This class encapsulates a thread-safe navigable map of LogSegment instances and provides the
@@ -141,7 +140,7 @@ public int numberOfSegments() {
      * @return the base offsets of all segments
      */
     public Collection<Long> baseOffsets() {
-        return values().stream().map(LogSegment::baseOffset).collect(Collectors.toList());
+        return values().stream().map(LogSegment::baseOffset).toList();
     }
 
     /**
@@ -182,7 +181,7 @@ public Collection<LogSegment> values() {
     public Collection<LogSegment> values(long from, long to) {
         if (from == to) {
             // Handle non-segment-aligned empty sets
-            return Collections.emptyList();
+            return List.of();
         } else if (to < from) {
             throw new IllegalArgumentException("Invalid log segment range: requested segments in " + topicPartition +
                     " from offset " + from + " which is greater than limit offset " + to);
@@ -197,7 +196,7 @@ public Collection<LogSegment> values(long from, long to) {
     public Collection<LogSegment> nonActiveLogSegmentsFrom(long from) {
         LogSegment activeSegment = lastSegment().get();
         if (from > activeSegment.baseOffset())
-            return Collections.emptyList();
+            return List.of();
         else
             return values(from, activeSegment.baseOffset());
     }
@@ -314,7 +313,7 @@ public Collection<LogSegment> higherSegments(long baseOffset) {
         Long higherOffset = segments.higherKey(baseOffset);
         if (higherOffset != null)
             return segments.tailMap(higherOffset, true).values();
-        return Collections.emptyList();
+        return List.of();
     }
 
     /**
@@ -334,7 +333,7 @@ public long sizeInBytes() {
      * @param predicate the predicate to be used for filtering segments.
      */
     public Collection<LogSegment> filter(Predicate<LogSegment> predicate) {
-        return values().stream().filter(predicate).collect(Collectors.toList());
+        return values().stream().filter(predicate).toList();
     }
 
     /**