KAFKA-16907: Fix RaftUtil's type complexity

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

@@ -39,6 +39,7 @@
 import org.apache.kafka.common.requests.UpdateRaftVoterRequest;
 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;
 
@@ -156,7 +157,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/LeaderState.java

@@ -31,6 +31,7 @@
 import org.apache.kafka.raft.internals.BatchAccumulator;
 import org.apache.kafka.raft.internals.KafkaRaftMetrics;
 import org.apache.kafka.raft.internals.RemoveVoterHandlerState;
+import org.apache.kafka.raft.utils.DynamicReconfigRpc;
 import org.apache.kafka.server.common.KRaftVersion;
 
 import org.slf4j.Logger;
@@ -223,7 +224,7 @@ public void resetAddVoterHandlerState(
         addVoterHandlerState.ifPresent(
             handlerState -> handlerState
                 .future()
-                .complete(RaftUtil.addVoterResponse(error, message))
+                .complete(DynamicReconfigRpc.addVoterResponse(error, message))
         );
         addVoterHandlerState = state;
         updateUncommittedVoterChangeMetric();
@@ -241,7 +242,7 @@ public void resetRemoveVoterHandlerState(
         removeVoterHandlerState.ifPresent(
             handlerState -> handlerState
                 .future()
-                .complete(RaftUtil.removeVoterResponse(error, message))
+                .complete(DynamicReconfigRpc.removeVoterResponse(error, message))
         );
         removeVoterHandlerState = state;
         updateUncommittedVoterChangeMetric();
@@ -386,6 +387,7 @@ public void appendStartOfEpochControlRecords(VoterSet.VoterNode localVoterNode,
                 return builder.build();
             }
         });
+        accumulator.forceDrain();
     }
 
     public long appendVotersRecord(VoterSet voters, long currentTimeMs) {