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

Code before:
                .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)
            .build();
        int resignLeadershipTimeout = context.checkQuorumTimeoutMs;

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


Review comment (suggestion):
let's not say sleep time - `// leader should send BeginQuorumEpochRequest to a node if beginQuorumEpochTimeoutMs time has elapsed without receiving a fetch request from that node`

Code after:
                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);
        Set<Integer> voters = Set.of(localId, remoteKey1.id(), remoteKey2.id());

        RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
            .withKip853Rpc(withKip853Rpc)
            .build();
        int resignLeadershipTimeout = context.checkQuorumTimeoutMs;