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

Code before:
        InetSocketAddress newAddress = InetSocketAddress.createUnresolved(
            "localhost",
            9990 + newVoter.id()
        );
        Endpoints newListeners = Endpoints.fromInetSocketAddresses(
            Map.of(context.channel.listenerName(), newAddress)
        );

        prepareToSendAddVoter(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 prepareToSendAddVoter(
        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)
        );
        context.pollUntilResponse();
        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(leader.id()));
        checkLeaderMetricValues(2, 1, 0, context);
    }


Review comment (refactor):
this name is a bit deceiving, this is all prep we do on the local (leader) node in order for it to be able to respond favorably to an addvoter request, could we perhaps rename to "prepareLeaderToReceiveAddVoter"? 

I'm also not sure how I feel about having a helper method for this, I do see how this is quite a bit of code duplication but I wonder if it might be more clear to have this written out explicitly in the tests (you probably don't need to keep all the same assertions for all the tests)

Code after:
        );
        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);

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

        RaftClientTestContext context = new RaftClientTestContext.Builder(local.id(), local.directoryId().get())