Repository: apache/kafka
File: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java (Java)

Code before:
        ReplicaKey replicaKey1 = replicaKey(remoteId1, withKip853Rpc);
        ReplicaKey replicaKey2 = replicaKey(remoteId2, withKip853Rpc);

        RaftClientTestContext context = new RaftClientTestContext.Builder(
                localId.id(), localId.directoryId().get())
                .withKip853Rpc(withKip853Rpc)
                .withStartingVoters(VoterSetTest.voterSet(Stream.of(localId, replicaKey1, replicaKey2)), KRaftVersion.KRAFT_VERSION_1)
                .build();

        context.unattachedToLeader();
        int epoch = context.currentEpoch();
        assertEquals(OptionalInt.of(localId.id()), context.currentLeader());

        // begin epoch requests sent out every beginQuorumEpochTimeoutMs if replicas have not fetched
        context.time.sleep(context.beginQuorumEpochTimeoutMs);
        context.client.poll();
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId1, remoteId2));

        long partialDelay = context.beginQuorumEpochTimeoutMs / 3;
        context.time.sleep(context.beginQuorumEpochTimeoutMs / 3);
        context.deliverRequest(context.fetchRequest(epoch, replicaKey1, 0, 0, 0));
        context.pollUntilResponse();

        context.time.sleep(context.beginQuorumEpochTimeoutMs - partialDelay);
        context.client.poll();
        // don't send BeginQuorumEpochRequest again for replicaKey1 since fetchRequest is sent.
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId2));

        context.deliverRequest(context.fetchRequest(epoch, replicaKey1, 0, 0, 0));
        context.pollUntilResponse();
        context.time.sleep(context.beginQuorumEpochTimeoutMs);
        context.client.poll();
        // should send BeginQuorumEpochRequest if sleep time equals beginQuorumEpochTimeoutMs
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId1, remoteId2));
    }


    @ParameterizedTest
    @ValueSource(booleans = { true, false })
    public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters(boolean withKip853Rpc) throws Exception {
        int localId = randomReplicaId();
        int remoteId1 = localId + 1;
        int remoteId2 = localId + 2;
        int observerId = localId + 3;
        ReplicaKey remoteKey1 = replicaKey(remoteId1, withKip853Rpc);
        ReplicaKey remoteKey2 = replicaKey(remoteId2, withKip853Rpc);
        ReplicaKey observerKey3 = replicaKey(observerId, withKip853Rpc);
        Set<Integer> voters = Set.of(localId, remoteKey1.id(), remoteKey2.id());

        RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
            .withKip853Rpc(withKip853Rpc)

Review comment (suggestion):
suggestion: `// leader will not send BeginQuorumEpochRequest again for replica 1 since fetchRequest was received before beginQuorumEpochTimeoutMs time has elapsed`

Code after:
        ReplicaKey replicaKey1 = replicaKey(remoteId1, withKip853Rpc);
        ReplicaKey replicaKey2 = replicaKey(remoteId2, withKip853Rpc);

        RaftClientTestContext context = new RaftClientTestContext.Builder(localKey.id(), localKey.directoryId().get())
            .withKip853Rpc(withKip853Rpc)
            .withStartingVoters(
                VoterSetTest.voterSet(Stream.of(localKey, replicaKey1, replicaKey2)),
                KRaftVersion.KRAFT_VERSION_1
            )
            .build();

        context.unattachedToLeader();
        int epoch = context.currentEpoch();
        assertEquals(OptionalInt.of(localKey.id()), context.currentLeader());

        // begin epoch requests sent out every beginQuorumEpochTimeoutMs if replicas have not fetched
        context.time.sleep(context.beginQuorumEpochTimeoutMs);
        context.client.poll();
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId1, remoteId2));

        long partialDelay = context.beginQuorumEpochTimeoutMs / 3;
        context.time.sleep(context.beginQuorumEpochTimeoutMs / 3);
        context.deliverRequest(context.fetchRequest(epoch, replicaKey1, 0, 0, 0));
        context.pollUntilResponse();

        context.time.sleep(context.beginQuorumEpochTimeoutMs - partialDelay);
        context.client.poll();
        // leader will not send BeginQuorumEpochRequest again for replica 1 since fetchRequest was received
        // before beginQuorumEpochTimeoutMs time has elapsed
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId2));

        context.deliverRequest(context.fetchRequest(epoch, replicaKey1, 0, 0, 0));
        context.pollUntilResponse();
        context.time.sleep(context.beginQuorumEpochTimeoutMs);
        context.client.poll();
        // leader should send BeginQuorumEpochRequest to a node if beginQuorumEpochTimeoutMs time has elapsed
        // without receiving a fetch request from that node
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId1, remoteId2));
    }


    @ParameterizedTest
    @ValueSource(booleans = { true, false })
    public void testLeaderShouldResignLeadershipIfNotGetFetchRequestFromMajorityVoters(boolean withKip853Rpc) throws Exception {
        int localId = randomReplicaId();
        int remoteId1 = localId + 1;
        int remoteId2 = localId + 2;
        int observerId = localId + 3;
        ReplicaKey remoteKey1 = replicaKey(remoteId1, withKip853Rpc);
        ReplicaKey remoteKey2 = replicaKey(remoteId2, withKip853Rpc);
        ReplicaKey observerKey3 = replicaKey(observerId, withKip853Rpc);