Repository: apache/kafka
File: clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/ReconfigurableQuorumIntegrationTest.java (Java)

Code before:
import org.junit.jupiter.api.Test;

import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

import static org.junit.jupiter.api.Assertions.assertEquals;

@Tag("integration")
public class ReconfigurableQuorumIntegrationTest {

    static Map<Integer, Uuid> descVoterDirs(Admin admin) throws ExecutionException, InterruptedException {
        var quorumInfo = admin.describeMetadataQuorum().quorumInfo().get();
        return quorumInfo.voters().stream().collect(Collectors.toMap(QuorumInfo.ReplicaState::replicaId, QuorumInfo.ReplicaState::replicaDirectoryId));
    }

    @Test
    public void testRemoveAndAddVoterWithValidClusterId() throws Exception {
        try (var cluster = new KafkaClusterTestKit.Builder(
                new TestKitNodes.Builder()
                        .setClusterId("test-cluster")
                        .setNumBrokerNodes(1)
                        .setNumControllerNodes(3)
                        .setFeature(KRaftVersion.FEATURE_NAME, KRaftVersion.KRAFT_VERSION_1.featureLevel())
                        .build()).build()
        ) {
            cluster.format();
            cluster.startup();
            try (Admin admin = Admin.create(cluster.clientProperties())) {
                TestUtils.waitForCondition(() -> {
                    Map<Integer, Uuid> voters = descVoterDirs(admin);
                    assertEquals(Set.of(3000, 3001, 3002), voters.keySet());
                    return voters.values().stream().noneMatch(directory -> directory.equals(Uuid.ZERO_UUID));
                }, "Initial quorum voters should be {3000, 3001, 3002} and all should have non-zero directory IDs");

                Uuid dirId = cluster.nodes().controllerNodes().get(3000).metadataDirectoryId();
                admin.removeRaftVoter(
                        3000,
                        dirId,
                        new RemoveRaftVoterOptions().setClusterId(Optional.of("test-cluster"))
                ).all().get();
                TestUtils.waitForCondition(() -> {
                    Map<Integer, Uuid> voters = descVoterDirs(admin);
                    assertEquals(Set.of(3001, 3002), voters.keySet());
                    return voters.values().stream().noneMatch(directory -> directory.equals(Uuid.ZERO_UUID));
                }, "After removing voter 3000, remaining voters should be {3001, 3002} with non-zero directory IDs");

                admin.addRaftVoter(
                        3000,

Review comment (suggestion):
`setFeature` is removed and I guess it is not necessary now.

Code after:

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

import static org.junit.jupiter.api.Assertions.assertEquals;

@Tag("integration")
public class ReconfigurableQuorumIntegrationTest {

    static Map<Integer, Uuid> descVoterDirs(Admin admin) throws ExecutionException, InterruptedException {
        var quorumInfo = admin.describeMetadataQuorum().quorumInfo().get();
        return quorumInfo.voters().stream().collect(Collectors.toMap(QuorumInfo.ReplicaState::replicaId, QuorumInfo.ReplicaState::replicaDirectoryId));
    }

    @Test
    public void testRemoveAndAddVoterWithValidClusterId() throws Exception {
        final var nodes = new TestKitNodes.Builder()
                .setClusterId("test-cluster")
                .setNumBrokerNodes(1)
                .setNumControllerNodes(3)
                .build();

        final Map<Integer, Uuid> initialVoters = new HashMap<>();
        for (final var controllerNode : nodes.controllerNodes().values()) {
            initialVoters.put(
                    controllerNode.id(),
                    controllerNode.metadataDirectoryId()
            );
        }

        try (var cluster = new KafkaClusterTestKit.Builder(nodes).setInitialVoterSet(initialVoters).build()) {
            cluster.format();
            cluster.startup();
            try (Admin admin = Admin.create(cluster.clientProperties())) {
                TestUtils.waitForCondition(() -> {
                    Map<Integer, Uuid> voters = descVoterDirs(admin);
                    assertEquals(Set.of(3000, 3001, 3002), voters.keySet());
                    return voters.values().stream().noneMatch(directory -> directory.equals(Uuid.ZERO_UUID));
                }, "Initial quorum voters should be {3000, 3001, 3002} and all should have non-zero directory IDs");

                Uuid dirId = cluster.nodes().controllerNodes().get(3000).metadataDirectoryId();
                admin.removeRaftVoter(
                        3000,
                        dirId,
                        new RemoveRaftVoterOptions().setClusterId(Optional.of("test-cluster"))
                ).all().get();
                TestUtils.waitForCondition(() -> {