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

Code before:
            .withKip853Rpc(withKip853Rpc)
            .build();

        context.unattachedToLeader();
        int epoch = context.currentEpoch();
        assertEquals(OptionalInt.of(localId), 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));

        int partialDelay = context.beginQuorumEpochTimeoutMs / 2;
        context.time.sleep(partialDelay);
        context.client.poll();
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of());

        context.time.sleep(context.beginQuorumEpochTimeoutMs - partialDelay);
        context.client.poll();
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId1, remoteId2));
    }

    @ParameterizedTest
    @ValueSource(booleans = { true, false })
    public void testBeginQuorumShouldNotSendAfterFetchRequest(boolean withKip853Rpc) throws Exception {
        ReplicaKey localId = replicaKey(randomReplicaId(), true);
        int remoteId1 = localId.id() + 1;
        int remoteId2 = localId.id() + 2;
        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();


Review comment (suggestion):
Make this name consistent with the rest of the names. This is the key and not the id. E.g. `localKey`.

Code after:
            .withKip853Rpc(withKip853Rpc)
            .build();

        context.unattachedToLeader();
        int epoch = context.currentEpoch();
        assertEquals(OptionalInt.of(localId), 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));

        int partialDelay = context.beginQuorumEpochTimeoutMs / 2;
        context.time.sleep(partialDelay);
        context.client.poll();
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of());

        context.time.sleep(context.beginQuorumEpochTimeoutMs - partialDelay);
        context.client.poll();
        context.assertSentBeginQuorumEpochRequest(epoch, Set.of(remoteId1, remoteId2));
    }

    @ParameterizedTest
    @ValueSource(booleans = { true, false })
    public void testBeginQuorumShouldNotSendAfterFetchRequest(boolean withKip853Rpc) throws Exception {
        ReplicaKey localKey = replicaKey(randomReplicaId(), true);
        int remoteId1 = localKey.id() + 1;
        int remoteId2 = localKey.id() + 2;
        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));