diff --git a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java index 3a32834de7f70..2a503afe2c695 100644 --- a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java +++ b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java @@ -3082,7 +3082,7 @@ private long maybeSendFetchOrFetchSnapshot(FollowerState state, long currentTime } private UpdateRaftVoterRequestData buildUpdateVoterRequest() { - return RaftUtil.updateVoterRequest( + return VoteRpc.updateVoterRequest( clusterId, quorum.localReplicaKeyOrThrow(), quorum.epoch(), diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java b/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java index 4a686d71ab3f2..a65b23dd198e9 100644 --- a/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java +++ b/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java @@ -204,7 +204,7 @@ public CompletableFuture handleUpdateVoterRequest( // Reply immediately and don't wait for the change to commit return CompletableFuture.completedFuture( - RaftUtil.updateVoterResponse( + VoteRpc.updateVoterResponse( Errors.NONE, requestListenerName, new LeaderAndEpoch( diff --git a/raft/src/main/java/org/apache/kafka/raft/utils/ApiMessageUtils.java b/raft/src/main/java/org/apache/kafka/raft/utils/ApiMessageUtils.java index 99ca74ea132ae..e4bac676ca80b 100644 --- a/raft/src/main/java/org/apache/kafka/raft/utils/ApiMessageUtils.java +++ b/raft/src/main/java/org/apache/kafka/raft/utils/ApiMessageUtils.java @@ -16,10 +16,12 @@ */ package org.apache.kafka.raft.utils; +import org.apache.kafka.common.message.ApiVersionsResponseData; import org.apache.kafka.common.message.BeginQuorumEpochResponseData; import org.apache.kafka.common.message.EndQuorumEpochResponseData; import org.apache.kafka.common.message.FetchResponseData; import org.apache.kafka.common.message.FetchSnapshotResponseData; +import org.apache.kafka.common.message.UpdateRaftVoterResponseData; import org.apache.kafka.common.message.VoteResponseData; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.ApiMessage; @@ -38,6 +40,10 @@ public static ApiMessage parseErrorResponse(ApiKeys apiKey, Errors error) { return new FetchResponseData().setErrorCode(error.code()); case FETCH_SNAPSHOT: return new FetchSnapshotResponseData().setErrorCode(error.code()); + case API_VERSIONS: + return new ApiVersionsResponseData().setErrorCode(error.code()); + case UPDATE_RAFT_VOTER: + return new UpdateRaftVoterResponseData().setErrorCode(error.code()); default: throw new IllegalArgumentException("Received response for unexpected request type: " + apiKey); } diff --git a/raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java b/raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java index 9b5dc05a003d3..a67d636e4504f 100644 --- a/raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java +++ b/raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java @@ -1758,7 +1758,7 @@ UpdateRaftVoterResponseData updateVoterResponse( Errors error, LeaderAndEpoch leaderAndEpoch ) { - return RaftUtil.updateVoterResponse( + return VoteRpc.updateVoterResponse( error, channel.listenerName(), leaderAndEpoch,