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

Code before:

        context.unattachedToLeader();
        int epoch = context.currentEpoch();

        ReplicaKey newVoter = replicaKey(local.id() + 2, true);
        InetSocketAddress newAddress = InetSocketAddress.createUnresolved(
            "localhost",
            9990 + newVoter.id()
        );
        Endpoints newListeners = Endpoints.fromInetSocketAddresses(
            Map.of(context.channel.listenerName(), newAddress)
        );

        prepareLeaderToReceiveAddVoter(context, epoch, local, follower, newVoter);

        // Attempt to add new voter to the quorum
        assertThrows(
            UnsupportedVersionException.class,
            () -> context.deliverRequest(
                context.addVoterRequest(
                    Integer.MAX_VALUE,
                    newVoter,
                    newListeners
                ).setAckWhenCommitted(false)
            )
        );
    }

    // This method sets up the context so a test can send an AddVoter request after
    // exiting this method
    private void prepareLeaderToReceiveAddVoter(
        RaftClientTestContext context,
        int epoch,
        ReplicaKey leader,
        ReplicaKey follower,
        ReplicaKey observer
    ) throws Exception {
        // Show that the observer is not currently a voter
        assertFalse(context.client.quorum().isVoter(observer));

        // Establish a HWM and fence previous leaders
        context.deliverRequest(
            context.fetchRequest(epoch, follower, context.log.endOffset().offset(), epoch, 0)
        );
        context.pollUntilResponse();
        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()));

        // Catch up the new voter to the leader's LEO, the new voter is still an observer at this point
        context.deliverRequest(
            context.fetchRequest(epoch, observer, context.log.endOffset().offset(), epoch, 0)
        );

Review comment (suggestion):
This is not testing `KafkaRaftClient`. This is testing the serialization code. It is testing that serializing a false "ack when committed" throws an exception when the target version is 0.

If you want to test this case for KafkaRaftClient, you need to make the local replica a non-voter that can auto-join and show that AddVoterRequest always sets "ack when committed" to false.

Code after:
                context.log.endOffset().offset(),
                epoch,
                0
            )
        );
        context.pollUntilResponse();
        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()));
        checkLeaderMetricValues(3, 0, 0, context);
    }

    @Test
    void testAddVoterInvalidClusterId() throws Exception {
        ReplicaKey local = replicaKey(randomReplicaId(), true);
        ReplicaKey follower = replicaKey(local.id() + 1, true);

        VoterSet voters = VoterSetTest.voterSet(Stream.of(local, follower));

        RaftClientTestContext context = new RaftClientTestContext.Builder(local.id(), local.directoryId().get())
            .withKip853Rpc(true)
            .withBootstrapSnapshot(Optional.of(voters))
            .withUnknownLeader(3)
            .build();

        context.unattachedToLeader();

        ReplicaKey newVoter = replicaKey(local.id() + 2, true);
        InetSocketAddress newAddress = InetSocketAddress.createUnresolved(
            "localhost",
            9990 + newVoter.id()
        );
        Endpoints newListeners = Endpoints.fromInetSocketAddresses(
            Map.of(context.channel.listenerName(), newAddress)
        );

        // empty cluster id is rejected
        context.deliverRequest(context.addVoterRequest("", Integer.MAX_VALUE, newVoter, newListeners));
        context.pollUntilResponse();
        context.assertSentAddVoterResponse(Errors.INCONSISTENT_CLUSTER_ID);

        // invalid cluster id is rejected
        context.deliverRequest(context.addVoterRequest("invalid-uuid", Integer.MAX_VALUE, newVoter, newListeners));
        context.pollUntilResponse();
        context.assertSentAddVoterResponse(Errors.INCONSISTENT_CLUSTER_ID);

        assertFalse(context.client.quorum().isVoter(newVoter));
    }

    @Test
    void testAddVoterToNotLeader() throws Exception {
        ReplicaKey local = replicaKey(randomReplicaId(), true);
        ReplicaKey follower = replicaKey(local.id() + 1, true);