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

Code before:

        RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
            .withUnknownLeader(epoch)
            .withKip853Rpc(withKip853Rpc)
            .build();

        context.pollUntilRequest();
        RaftRequest.Outbound request = context.assertSentFetchRequest(epoch, 0L, 0, OptionalLong.empty());
        assertTrue(context.client.quorum().isUnattached());
        assertTrue(context.client.quorum().isVoter());

        // receives a fetch response specifying who the leader is
        Errors responseError = (request.destination().id() == otherNodeId) ? Errors.NOT_LEADER_OR_FOLLOWER : Errors.NONE;
        context.deliverResponse(
            request.correlationId(),
            request.destination(),
            context.fetchResponse(epoch, leaderNodeId, MemoryRecords.EMPTY, 0L, responseError)
        );

        context.client.poll();
        assertTrue(context.client.quorum().isFollower());
    }

    @ParameterizedTest
    @ValueSource(booleans = { true, false })
    public void testUnattachedWithLeaderCanBecomeFollowerAfterFindingLeader(boolean withKip853Rpc) throws Exception {
        int localId = randomReplicaId();
        int otherNodeId = localId + 1;
        int leaderNodeId = localId + 2;
        int epoch = 5;
        Set<Integer> voters = Set.of(localId, otherNodeId, leaderNodeId);
        List<InetSocketAddress> bootstrapServers = voters
            .stream()
            .map(RaftClientTestContext::mockAddress)
            .toList();

        RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
            .withBootstrapServers(Optional.of(bootstrapServers))
            .withElectedLeader(epoch, leaderNodeId)
            .withKip853Rpc(withKip853Rpc)
            .build();

        // after fetch timeout, node will become prospective with leader
        context.time.sleep(context.fetchTimeoutMs);
        context.pollUntilRequest();
        context.assertSentPreVoteRequest(epoch, 0, 0L, 2);
        context.assertElectedLeader(epoch, leaderNodeId);

        // after election loss node will become follower because it had a last known leader
        context.time.sleep(context.electionTimeoutMs() * 2L);
        context.pollUntilRequest();

Review comment (bug):
this name was incorrect and should have been something like `testProspectiveWithLeaderCanBecomeFollower` - but in that case it already duplicates a test in KafkaRaftClientPreVoteTest so I'm deleting this

Code after:

        RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
            .withUnknownLeader(epoch)
            .withKip853Rpc(withKip853Rpc)
            .build();

        context.pollUntilRequest();
        RaftRequest.Outbound request = context.assertSentFetchRequest(epoch, 0L, 0, OptionalLong.empty());
        assertTrue(context.client.quorum().isUnattached());
        assertTrue(context.client.quorum().isVoter());

        // receives a fetch response specifying who the leader is
        Errors responseError = (request.destination().id() == otherNodeId) ? Errors.NOT_LEADER_OR_FOLLOWER : Errors.NONE;
        context.deliverResponse(
            request.correlationId(),
            request.destination(),
            context.fetchResponse(epoch, leaderNodeId, MemoryRecords.EMPTY, 0L, responseError)
        );

        context.client.poll();
        assertTrue(context.client.quorum().isFollower());
    }

    @ParameterizedTest
    @ValueSource(booleans = { true, false })
    public void testInitializeObserverNoPreviousState(boolean withKip853Rpc) throws Exception {
        int localId = randomReplicaId();
        int leaderId = localId + 1;
        int otherNodeId = localId + 2;
        int epoch = 5;
        Set<Integer> voters = Set.of(leaderId, otherNodeId);

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

        context.pollUntilRequest();
        RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
        assertTrue(voters.contains(fetchRequest.destination().id()));
        context.assertFetchRequestData(fetchRequest, 0, 0L, 0, context.client.highWatermark());

        context.deliverResponse(
            fetchRequest.correlationId(),
            fetchRequest.destination(),
            context.fetchResponse(epoch, leaderId, MemoryRecords.EMPTY, 0L, Errors.FENCED_LEADER_EPOCH)
        );

        context.client.poll();
        context.assertElectedLeader(epoch, leaderId);
    }