MINOR: Cleanup Raft Module (apache#19284)

Given that now we support Java 17 on our brokers, this PR replace the
use of the following in raft 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

raft/src/main/java/org/apache/kafka/raft/Batch.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.raft;
 
-import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
@@ -181,7 +180,7 @@ public static <T> Batch<T> control(
             appendTimestamp,
             sizeInBytes,
             baseOffset + records.size() - 1,
-            Collections.emptyList(),
+            List.of(),
             records
         );
     }
@@ -219,7 +218,7 @@ public static <T> Batch<T> data(
             sizeInBytes,
             baseOffset + records.size() - 1,
             records,
-            Collections.emptyList()
+            List.of()
         );
     }
 }

raft/src/main/java/org/apache/kafka/raft/ControlRecord.java

@@ -79,23 +79,6 @@ public ControlRecordType type() {
         return recordType;
     }
 
-    public short version() {
-        switch (recordType) {
-            case LEADER_CHANGE:
-                return ((LeaderChangeMessage) message).version();
-            case SNAPSHOT_HEADER:
-                return ((SnapshotHeaderRecord) message).version();
-            case SNAPSHOT_FOOTER:
-                return ((SnapshotFooterRecord) message).version();
-            case KRAFT_VERSION:
-                return ((KRaftVersionRecord) message).version();
-            case KRAFT_VOTERS:
-                return ((VotersRecord) message).version();
-            default:
-                throw new IllegalStateException(String.format("Unknown control record type %s", recordType));
-        }
-    }
-
     public ApiMessage message() {
         return message;
     }

raft/src/main/java/org/apache/kafka/raft/DynamicVoter.java

@@ -22,7 +22,7 @@
 import org.apache.kafka.common.network.ListenerName;
 
 import java.net.InetSocketAddress;
-import java.util.Collections;
+import java.util.Map;
 import java.util.Objects;
 
 /**
@@ -153,7 +153,7 @@ public int port() {
 
     public VoterSet.VoterNode toVoterNode(String controllerListenerName) {
         ReplicaKey voterKey = ReplicaKey.of(nodeId, directoryId);
-        Endpoints listeners = Endpoints.fromInetSocketAddresses(Collections.singletonMap(
+        Endpoints listeners = Endpoints.fromInetSocketAddresses(Map.of(
                 ListenerName.normalised(controllerListenerName),
                 new InetSocketAddress(host, port)));
         SupportedVersionRange supportedKRaftVersion =

raft/src/main/java/org/apache/kafka/raft/ElectionState.java

@@ -63,7 +63,7 @@ public boolean isLeader(int nodeId) {
     /**
      * Return if the replica has voted for the given candidate.
      *
-     * A replica has voted for a candidate if all of the following are true:
+     * A replica has voted for a candidate if all the following are true:
      * 1. the node's id and voted id match and
      * 2. if the voted directory id is set, it matches the node's directory id
      *

raft/src/main/java/org/apache/kafka/raft/Endpoints.java

@@ -31,7 +31,6 @@
 
 import java.net.InetSocketAddress;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
@@ -147,7 +146,7 @@ public UpdateRaftVoterRequestData.ListenerCollection toUpdateVoterRequest() {
         return listeners;
     }
 
-    private static final Endpoints EMPTY = new Endpoints(Collections.emptyMap());
+    private static final Endpoints EMPTY = new Endpoints(Map.of());
     public static Endpoints empty() {
         return EMPTY;
     }
@@ -188,7 +187,7 @@ public static Endpoints fromBeginQuorumEpochResponse(
         return Optional.ofNullable(endpoints.find(leaderId))
             .map(endpoint ->
                 new Endpoints(
-                    Collections.singletonMap(
+                    Map.of(
                         listenerName,
                         InetSocketAddress.createUnresolved(endpoint.host(), endpoint.port())
                     )
@@ -217,7 +216,7 @@ public static Endpoints fromEndQuorumEpochResponse(
         return Optional.ofNullable(endpoints.find(leaderId))
             .map(endpoint ->
                 new Endpoints(
-                    Collections.singletonMap(
+                    Map.of(
                         listenerName,
                         InetSocketAddress.createUnresolved(endpoint.host(), endpoint.port())
                     )
@@ -234,7 +233,7 @@ public static Endpoints fromVoteResponse(
         return Optional.ofNullable(endpoints.find(leaderId))
             .map(endpoint ->
                 new Endpoints(
-                    Collections.singletonMap(
+                    Map.of(
                         listenerName,
                         InetSocketAddress.createUnresolved(endpoint.host(), endpoint.port())
                     )
@@ -251,7 +250,7 @@ public static Endpoints fromFetchResponse(
         return Optional.ofNullable(endpoints.find(leaderId))
             .map(endpoint ->
                 new Endpoints(
-                    Collections.singletonMap(
+                    Map.of(
                         listenerName,
                         InetSocketAddress.createUnresolved(endpoint.host(), endpoint.port())
                     )
@@ -268,7 +267,7 @@ public static Endpoints fromFetchSnapshotResponse(
         return Optional.ofNullable(endpoints.find(leaderId))
             .map(endpoint ->
                 new Endpoints(
-                    Collections.singletonMap(
+                    Map.of(
                         listenerName,
                         InetSocketAddress.createUnresolved(endpoint.host(), endpoint.port())
                     )

raft/src/main/java/org/apache/kafka/raft/KafkaNetworkChannel.java

@@ -140,7 +140,7 @@ private void sendOnComplete(RaftRequest.Outbound request, ClientResponse clientR
             log.error("Request {} failed due to unsupported version error", request, clientResponse.versionMismatch());
             response = errorResponse(request.data(), Errors.UNSUPPORTED_VERSION);
         } else if (clientResponse.authenticationException() != null) {
-            // For now we treat authentication errors as retriable. We use the
+            // For now, we treat authentication errors as retriable. We use the
             // `NETWORK_EXCEPTION` error code for lack of a good alternative.
             // Note that `NodeToControllerChannelManager` will still log the
             // authentication errors so that users have a chance to fix the problem.

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

@@ -98,7 +98,6 @@
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HexFormat;
 import java.util.IdentityHashMap;
 import java.util.Iterator;
@@ -593,9 +592,7 @@ public void initialize(
         this.updateVoterHandler = new UpdateVoterHandler(
             nodeId,
             partitionState,
-            channel.listenerName(),
-            time,
-            quorumConfig.requestTimeoutMs()
+            channel.listenerName()
         );
     }
 
@@ -654,7 +651,7 @@ private void onBecomeLeader(long currentTimeMs) {
         log.initializeLeaderEpoch(quorum.epoch());
 
         // The high watermark can only be advanced once we have written a record
-        // from the new leader's epoch. Hence we write a control message immediately
+        // from the new leader's epoch. Hence, we write a control message immediately
         // to ensure there is no delay committing pending data.
         state.appendStartOfEpochControlRecords(quorum.localVoterNodeOrThrow(), currentTimeMs);
 
@@ -1900,7 +1897,7 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
      * - {@link Errors#FENCED_LEADER_EPOCH} if the epoch is smaller than this node's epoch
      * - {@link Errors#INVALID_REQUEST} if the request epoch is larger than the leader's current epoch
      *     or if either the fetch offset or the last fetched epoch is invalid
-     * - {@link Errors#SNAPSHOT_NOT_FOUND} if the request snapshot id does not exists
+     * - {@link Errors#SNAPSHOT_NOT_FOUND} if the request snapshot id does not exist
      * - {@link Errors#POSITION_OUT_OF_RANGE} if the request snapshot offset out of range
      */
     private FetchSnapshotResponseData handleFetchSnapshotRequest(
@@ -2418,7 +2415,7 @@ private boolean handleUpdateVoterResponse(
         final Endpoints leaderEndpoints;
         if (responseLeaderId.isPresent() && data.currentLeader().host().isEmpty()) {
             leaderEndpoints = Endpoints.fromInetSocketAddresses(
-                Collections.singletonMap(
+                Map.of(
                     channel.listenerName(),
                     InetSocketAddress.createUnresolved(
                         data.currentLeader().host(),
@@ -2887,15 +2884,11 @@ private FetchRequestData buildFetchRequest() {
 
     private long maybeSendFetchToAnyBootstrap(long currentTimeMs) {
         Optional<Node> readyNode = requestManager.findReadyBootstrapServer(currentTimeMs);
-        if (readyNode.isPresent()) {
-            return maybeSendRequest(
-                currentTimeMs,
-                readyNode.get(),
-                this::buildFetchRequest
-            );
-        } else {
-            return requestManager.backoffBeforeAvailableBootstrapServer(currentTimeMs);
-        }
+        return readyNode.map(node -> maybeSendRequest(
+            currentTimeMs,
+            node,
+            this::buildFetchRequest
+        )).orElseGet(() -> requestManager.backoffBeforeAvailableBootstrapServer(currentTimeMs));
     }
 
     private FetchSnapshotRequestData buildFetchSnapshotRequest(OffsetAndEpoch snapshotId, long snapshotSize) {
@@ -3331,7 +3324,7 @@ private long pollCurrentState(long currentTimeMs) {
     }
 
     private void pollListeners() {
-        // Apply all of the pending registration
+        // Apply all the pending registration
         while (true) {
             Registration<T> registration = pendingRegistrations.poll();
             if (registration == null) {
@@ -3825,7 +3818,7 @@ private void fireHandleCommit(long baseOffset, Records records) {
          * This API is used for committed records originating from {@link #prepareAppend(int, List)}
          * on this instance. In this case, we are able to save the original record objects, which
          * saves the need to read them back from disk. This is a nice optimization for the leader
-         * which is typically doing more work than all of the * followers.
+         * which is typically doing more work than all the * followers.
          */
         private void fireHandleCommit(
             long baseOffset,
@@ -3835,7 +3828,7 @@ private void fireHandleCommit(
             List<T> records
         ) {
             Batch<T> batch = Batch.data(baseOffset, epoch, appendTimestamp, sizeInBytes, records);
-            MemoryBatchReader<T> reader = MemoryBatchReader.of(Collections.singletonList(batch), this);
+            MemoryBatchReader<T> reader = MemoryBatchReader.of(List.of(batch), this);
             fireHandleCommit(reader);
         }
 

raft/src/main/java/org/apache/kafka/raft/LeaderState.java

@@ -76,7 +76,7 @@ public class LeaderState<T> implements EpochState {
     private final Map<ReplicaKey, ReplicaState> observerStates = new HashMap<>();
     private final Logger log;
     private final BatchAccumulator<T> accumulator;
-    // The set includes all of the followers voters that FETCH or FETCH_SNAPSHOT during the current checkQuorumTimer interval.
+    // The set includes all the followers voters that FETCH or FETCH_SNAPSHOT during the current checkQuorumTimer interval.
     private final Set<Integer> fetchedVoters = new HashSet<>();
     private final Timer checkQuorumTimer;
     private final int checkQuorumTimeoutMs;

raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java

@@ -24,7 +24,6 @@
 import org.apache.kafka.common.utils.Utils;
 
 import java.net.InetSocketAddress;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -45,7 +44,7 @@
  * The default raft timeouts are relatively low compared to some other timeouts such as
  * request.timeout.ms. This is part of a general design philosophy where we see changing
  * the leader of a Raft cluster as a relatively quick operation. For example, the KIP-631
- * controller should be able to transition from standby to active without reloading all of
+ * controller should be able to transition from standby to active without reloading all
  * the metadata. The standby is a "hot" standby, not a "cold" one.
  */
 public class QuorumConfig {
@@ -59,14 +58,14 @@ public class QuorumConfig {
     public static final String QUORUM_VOTERS_DOC = "Map of id/endpoint information for " +
         "the set of voters in a comma-separated list of <code>{id}@{host}:{port}</code> entries. " +
         "For example: <code>1@localhost:9092,2@localhost:9093,3@localhost:9094</code>";
-    public static final List<String> DEFAULT_QUORUM_VOTERS = Collections.emptyList();
+    public static final List<String> DEFAULT_QUORUM_VOTERS = List.of();
 
     public static final String QUORUM_BOOTSTRAP_SERVERS_CONFIG = QUORUM_PREFIX + "bootstrap.servers";
     public static final String QUORUM_BOOTSTRAP_SERVERS_DOC = "List of endpoints to use for " +
         "bootstrapping the cluster metadata. The endpoints are specified in comma-separated list " +
         "of <code>{host}:{port}</code> entries. For example: " +
         "<code>localhost:9092,localhost:9093,localhost:9094</code>.";
-    public static final List<String> DEFAULT_QUORUM_BOOTSTRAP_SERVERS = Collections.emptyList();
+    public static final List<String> DEFAULT_QUORUM_BOOTSTRAP_SERVERS = List.of();
 
     public static final String QUORUM_ELECTION_TIMEOUT_MS_CONFIG = QUORUM_PREFIX + "election.timeout.ms";
     public static final String QUORUM_ELECTION_TIMEOUT_MS_DOC = "Maximum time in milliseconds to wait " +

raft/src/main/java/org/apache/kafka/raft/QuorumState.java

@@ -28,7 +28,6 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.OptionalInt;
@@ -185,7 +184,7 @@ public void initialize(OffsetAndEpoch logEndOffsetAndEpoch) throws IllegalStateE
                 election.epoch(),
                 partitionState.lastVoterSet().voterIds(),
                 randomElectionTimeoutMs(),
-                Collections.emptyList(),
+                List.of(),
                 localListeners,
                 logContext
             );