KAFKA-16907: Resolve the ClassDataAbstractionCoupling and ClassFanOutComplexity issue caused by RaftUtils

Author: frankvicky

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

@@ -35,6 +35,7 @@
 import org.apache.kafka.common.requests.FetchSnapshotRequest;
 import org.apache.kafka.common.requests.VoteRequest;
 import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.raft.utils.ApiMessageUtils;
 import org.apache.kafka.server.util.InterBrokerSendThread;
 import org.apache.kafka.server.util.RequestAndCompletionHandler;
 
@@ -152,7 +153,7 @@ private void sendOnComplete(RaftRequest.Outbound request, ClientResponse clientR
 
     private ApiMessage errorResponse(ApiMessage request, Errors error) {
         ApiKeys apiKey = ApiKeys.forId(request.apiKey());
-        return RaftUtil.errorResponse(apiKey, error);
+        return ApiMessageUtils.parseErrorResponse(apiKey, error);
     }
 
     @Override

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

@@ -34,6 +34,7 @@
 import org.apache.kafka.raft.internals.ReplicaKey;
 import org.apache.kafka.raft.internals.UpdateVoterHandlerState;
 import org.apache.kafka.raft.internals.VoterSet;
+import org.apache.kafka.raft.utils.VoteRpc;
 import org.apache.kafka.server.common.KRaftVersion;
 
 import org.slf4j.Logger;
@@ -225,7 +226,7 @@ public void resetAddVoterHandlerState(
         addVoterHandlerState.ifPresent(
             handlerState -> handlerState
                 .future()
-                .complete(RaftUtil.addVoterResponse(error, message))
+                .complete(VoteRpc.addVoterResponse(error, message))
         );
         addVoterHandlerState = state;
     }
@@ -242,7 +243,7 @@ public void resetRemoveVoterHandlerState(
         removeVoterHandlerState.ifPresent(
             handlerState -> handlerState
                 .future()
-                .complete(RaftUtil.removeVoterResponse(error, message))
+                .complete(VoteRpc.removeVoterResponse(error, message))
         );
         removeVoterHandlerState = state;
     }
@@ -719,7 +720,7 @@ private void updateVoterAndObserverStates(VoterSet lastVoterSet) {
         }
     }
 
-    static class ReplicaState implements Comparable<ReplicaState> {
+    public static class ReplicaState implements Comparable<ReplicaState> {
         private ReplicaKey replicaKey;
         private Endpoints listeners;
         private Optional<LogOffsetMetadata> endOffset;