KAFKA-18561: Remove withKip853Rpc and replace it with withRaftProtocol

raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientClusterAuthTest.java

@@ -38,14 +38,14 @@
 public class KafkaRaftClientClusterAuthTest {
     @ParameterizedTest
     @ValueSource(booleans = { true, false })
-    void testClusterAuthorizationFailedInFetch(boolean withKip853Rpc) throws Exception {
+    void testClusterAuthorizationFailedInFetch(boolean isReconfigSupported) throws Exception {
         int localId = randomReplicaId();
         int otherNodeId = localId + 1;
         int epoch = 5;
         Set<Integer> voters = Set.of(localId, otherNodeId);
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
-                .withKip853Rpc(withKip853Rpc)
+                .withRaftProtocol(isReconfigSupported ? RaftProtocol.KIP_853_PROTOCOL : RaftProtocol.KIP_595_PROTOCOL)
                 .withElectedLeader(epoch, otherNodeId)
                 .build();
 
@@ -66,7 +66,7 @@ void testClusterAuthorizationFailedInFetch(boolean withKip853Rpc) throws Excepti
 
     @ParameterizedTest
     @ValueSource(booleans = { true, false })
-    void testClusterAuthorizationFailedInBeginQuorumEpoch(boolean withKip853Rpc) throws Exception {
+    void testClusterAuthorizationFailedInBeginQuorumEpoch(boolean isReconfigSupported) throws Exception {
         int localId = randomReplicaId();
         int otherNodeId = localId + 1;
         int epoch = 5;
@@ -75,7 +75,7 @@ void testClusterAuthorizationFailedInBeginQuorumEpoch(boolean withKip853Rpc) thr
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
                 .updateRandom(r -> r.mockNextInt(DEFAULT_ELECTION_TIMEOUT_MS, 0))
                 .withUnknownLeader(epoch - 1)
-                .withKip853Rpc(withKip853Rpc)
+                .withRaftProtocol(isReconfigSupported ? RaftProtocol.KIP_853_PROTOCOL : RaftProtocol.KIP_595_PROTOCOL)
                 .build();
 
         context.time.sleep(context.electionTimeoutMs());
@@ -96,15 +96,15 @@ void testClusterAuthorizationFailedInBeginQuorumEpoch(boolean withKip853Rpc) thr
 
     @ParameterizedTest
     @ValueSource(booleans = { true, false })
-    void testClusterAuthorizationFailedInVote(boolean withKip853Rpc) throws Exception {
+    void testClusterAuthorizationFailedInVote(boolean isReconfigSupported) throws Exception {
         int localId = randomReplicaId();
         int otherNodeId = localId + 1;
         int epoch = 5;
         Set<Integer> voters = Set.of(localId, otherNodeId);
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
                 .withUnknownLeader(epoch - 1)
-                .withKip853Rpc(withKip853Rpc)
+                .withRaftProtocol(isReconfigSupported ? RaftProtocol.KIP_853_PROTOCOL : RaftProtocol.KIP_595_PROTOCOL)
                 .build();
 
         // Become a candidate
@@ -122,14 +122,14 @@ void testClusterAuthorizationFailedInVote(boolean withKip853Rpc) throws Exceptio
 
     @ParameterizedTest
     @ValueSource(booleans = { true, false })
-    void testClusterAuthorizationFailedInEndQuorumEpoch(boolean withKip853Rpc) throws Exception {
+    void testClusterAuthorizationFailedInEndQuorumEpoch(boolean isReconfigSupported) throws Exception {
         int localId = randomReplicaId();
         int otherNodeId = localId + 1;
         Set<Integer> voters = Set.of(localId, otherNodeId);
 
         RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
                 .withUnknownLeader(1)
-                .withKip853Rpc(withKip853Rpc)
+                .withRaftProtocol(isReconfigSupported ? RaftProtocol.KIP_853_PROTOCOL : RaftProtocol.KIP_595_PROTOCOL)
                 .build();
 
         context.unattachedToLeader();

raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientFetchTest.java

@@ -67,7 +67,7 @@ private static void testFetchResponseWithInvalidRecord(MemoryRecords records, in
                 VoterSetTest.voterSet(Stream.of(local, electedLeader)), KRaftVersion.KRAFT_VERSION_1
             )
             .withElectedLeader(epoch, electedLeader.id())
-            .withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_996_PROTOCOL)
+            .withRaftProtocol(RaftProtocol.KIP_996_PROTOCOL)
             .build();
 
         context.pollUntilRequest();
@@ -102,7 +102,7 @@ void testReplicationOfHigherPartitionLeaderEpoch() throws Exception {
                 VoterSetTest.voterSet(Stream.of(local, electedLeader)), KRaftVersion.KRAFT_VERSION_1
             )
             .withElectedLeader(epoch, electedLeader.id())
-            .withRaftProtocol(RaftClientTestContext.RaftProtocol.KIP_996_PROTOCOL)
+            .withRaftProtocol(RaftProtocol.KIP_996_PROTOCOL)
             .build();
 
         context.pollUntilRequest();

raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientPreVoteTest.java

@@ -21,7 +21,6 @@
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.record.MemoryRecords;
-import org.apache.kafka.raft.RaftClientTestContext.RaftProtocol;
 import org.apache.kafka.server.common.KRaftVersion;
 
 import org.junit.jupiter.params.ParameterizedTest;
@@ -35,7 +34,7 @@
 
 import static org.apache.kafka.raft.KafkaRaftClientTest.randomReplicaId;
 import static org.apache.kafka.raft.KafkaRaftClientTest.replicaKey;
-import static org.apache.kafka.raft.RaftClientTestContext.RaftProtocol.KIP_996_PROTOCOL;
+import static org.apache.kafka.raft.RaftProtocol.KIP_996_PROTOCOL;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;