[GitHub] [kafka] hachikuji commented on a change in pull request #10220: KAFKA-12383: Get RaftClusterTest.java and other KIP-500 junit tests working

[GitBox]

hachikuji commented on a change in pull request #10220:
URL: https://github.com/apache/kafka/pull/10220#discussion_r588522951



##########
File path: core/src/test/java/kafka/test/junit/ClusterTestExtensions.java
##########
@@ -199,12 +209,25 @@ private void processClusterTest(ClusterTest annot, 
ClusterTestDefaults defaults,
         }
 
         switch (type) {
-            case ZK:
-            case BOTH:
+            case ZK: {
                 ClusterConfig config = builder.build();
                 config.serverProperties().putAll(properties);
                 testInvocations.accept(new ZkClusterInvocationContext(config));
                 break;
+            } case RAFT: {
+                ClusterConfig config = builder.build();
+                config.serverProperties().putAll(properties);
+                testInvocations.accept(new 
RaftClusterInvocationContext(config));
+                break;
+            } case BOTH: {
+                ClusterConfig zkConfig = builder.build();

Review comment:
       nit: not a big deal, but I suspect there's a way to remove some of the 
duplication here in this switch. For example, perhaps we could add a method to 
`Type` to build the invocation context:
   ```java
     List<ClusterInvocationContext> invocationContexts(ClusterConfig config);
   ```

##########
File path: core/src/test/java/kafka/test/junit/RaftClusterInvocationContext.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.test.junit;
+
+import kafka.network.SocketServer;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.test.ClusterConfig;
+import kafka.test.ClusterInstance;
+import kafka.testkit.KafkaClusterTestKit;
+import kafka.testkit.TestKitNodes;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.metadata.BrokerState;
+import org.junit.jupiter.api.extension.AfterTestExecutionCallback;
+import org.junit.jupiter.api.extension.BeforeTestExecutionCallback;
+import org.junit.jupiter.api.extension.Extension;
+import org.junit.jupiter.api.extension.TestTemplateInvocationContext;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+/**
+ * Wraps a {@link KafkaClusterTestKit} inside lifecycle methods for a test 
invocation. Each instance of this
+ * class is provided with a configuration for the cluster.
+ *
+ * This context also provides parameter resolvers for:
+ *
+ * <ul>
+ *     <li>ClusterConfig (the same instance passed to the constructor)</li>
+ *     <li>ClusterInstance (includes methods to expose underlying 
SocketServer-s)</li>
+ *     <li>IntegrationTestHelper (helper methods)</li>
+ * </ul>
+ */
+public class RaftClusterInvocationContext implements 
TestTemplateInvocationContext {
+
+    private final ClusterConfig clusterConfig;
+    private final AtomicReference<KafkaClusterTestKit> clusterReference;
+
+    public RaftClusterInvocationContext(ClusterConfig clusterConfig) {
+        this.clusterConfig = clusterConfig;
+        this.clusterReference = new AtomicReference<>();
+    }
+
+    @Override
+    public String getDisplayName(int invocationIndex) {
+        String clusterDesc = clusterConfig.nameTags().entrySet().stream()
+            .map(Object::toString)
+            .collect(Collectors.joining(", "));
+        return String.format("[Quorum %d] %s", invocationIndex, clusterDesc);
+    }
+
+    @Override
+    public List<Extension> getAdditionalExtensions() {
+        return Arrays.asList(
+            (BeforeTestExecutionCallback) context -> {
+                KafkaClusterTestKit.Builder builder = new 
KafkaClusterTestKit.Builder(
+                    new TestKitNodes.Builder().
+                        setNumKip500BrokerNodes(clusterConfig.numBrokers()).
+                        
setNumControllerNodes(clusterConfig.numControllers()).build());
+
+                // Copy properties into the TestKit builder
+                clusterConfig.serverProperties().forEach((key, value) -> 
builder.setConfigProp(key.toString(), value.toString()));
+                // TODO how to pass down security protocol and listener name?
+                KafkaClusterTestKit cluster = builder.build();
+                clusterReference.set(cluster);
+                cluster.format();
+                cluster.startup();
+                kafka.utils.TestUtils.waitUntilTrue(
+                    () -> cluster.kip500Brokers().get(0).currentState() == 
BrokerState.RUNNING,
+                    () -> "Broker never made it to RUNNING state.",
+                    org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS,
+                    100L);
+            },
+            (AfterTestExecutionCallback) context -> 
clusterReference.get().close(),
+            new ClusterInstanceParameterResolver(new 
RaftClusterInstance(clusterReference, clusterConfig)),
+            new GenericParameterResolver<>(clusterConfig, ClusterConfig.class)
+        );
+    }
+
+    public static class RaftClusterInstance implements ClusterInstance {
+
+        private final AtomicReference<KafkaClusterTestKit> clusterReference;
+        private final ClusterConfig clusterConfig;
+        final AtomicBoolean started = new AtomicBoolean(false);
+        final AtomicBoolean stopped = new AtomicBoolean(false);
+
+        RaftClusterInstance(AtomicReference<KafkaClusterTestKit> 
clusterReference, ClusterConfig clusterConfig) {
+            this.clusterReference = clusterReference;
+            this.clusterConfig = clusterConfig;
+        }
+
+        @Override
+        public String bootstrapServers() {
+            return 
clusterReference.get().clientProperties().getProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
+        }
+
+        @Override
+        public Collection<SocketServer> brokerSocketServers() {
+            return clusterReference.get().kip500Brokers().values().stream()
+                .map(BrokerServer::socketServer)
+                .collect(Collectors.toList());
+        }
+
+        @Override
+        public ListenerName clientListener() {
+            return ListenerName.normalised("EXTERNAL");
+        }
+
+        @Override
+        public Collection<SocketServer> controllerSocketServers() {
+            return clusterReference.get().controllers().values().stream()
+                .map(ControllerServer::socketServer)
+                .collect(Collectors.toList());
+        }
+
+        @Override
+        public SocketServer anyBrokerSocketServer() {
+            return clusterReference.get().kip500Brokers().values().stream()
+                .map(BrokerServer::socketServer)
+                .findFirst()
+                .orElseThrow(() -> new RuntimeException("No broker 
SocketServers found"));
+        }
+
+        @Override
+        public SocketServer anyControllerSocketServer() {
+            return clusterReference.get().controllers().values().stream()
+                .map(ControllerServer::socketServer)
+                .findFirst()
+                .orElseThrow(() -> new RuntimeException("No controller 
SocketServers found"));
+        }
+
+        @Override
+        public ClusterType clusterType() {
+            return ClusterType.RAFT;
+        }
+
+        @Override
+        public ClusterConfig config() {
+            return clusterConfig;
+        }
+
+        @Override
+        public Object getUnderlying() {

Review comment:
       Is there any benefit letting this return the more specific 
`KafkaClusterTestKit` type?

##########
File path: core/src/test/java/kafka/testkit/Kip500BrokerNode.java
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class Kip500BrokerNode implements TestKitNode {
+    public static class Builder {
+        private int id = -1;
+        private Uuid incarnationId = null;
+        private String metadataDirectory = null;
+        private List<String> logDataDirectories = null;
+
+        public Builder setId(int id) {
+            this.id = id;
+            return this;
+        }
+
+        public Builder setLogDirectories(List<String> logDataDirectories) {
+            this.logDataDirectories = logDataDirectories;
+            return this;
+        }
+
+        public Kip500BrokerNode build() {
+            if (id == -1) {
+                throw new RuntimeException("You must set the node id");
+            }
+            if (incarnationId == null) {
+                incarnationId = Uuid.randomUuid();
+            }
+            if (metadataDirectory == null) {
+                metadataDirectory = String.format("kip500broker_%d_meta", id);

Review comment:
       nit: do we need the `kip500` prefix? We don't use it in 
`ControllerNode`. Similarly, maybe we can call this `BrokerNode` instead of 
`Kip500BrokerNode`?

##########
File path: core/src/test/java/kafka/test/junit/RaftClusterInvocationContext.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.test.junit;
+
+import kafka.network.SocketServer;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.test.ClusterConfig;
+import kafka.test.ClusterInstance;
+import kafka.testkit.KafkaClusterTestKit;
+import kafka.testkit.TestKitNodes;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.metadata.BrokerState;
+import org.junit.jupiter.api.extension.AfterTestExecutionCallback;
+import org.junit.jupiter.api.extension.BeforeTestExecutionCallback;
+import org.junit.jupiter.api.extension.Extension;
+import org.junit.jupiter.api.extension.TestTemplateInvocationContext;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+/**
+ * Wraps a {@link KafkaClusterTestKit} inside lifecycle methods for a test 
invocation. Each instance of this
+ * class is provided with a configuration for the cluster.
+ *
+ * This context also provides parameter resolvers for:
+ *
+ * <ul>
+ *     <li>ClusterConfig (the same instance passed to the constructor)</li>
+ *     <li>ClusterInstance (includes methods to expose underlying 
SocketServer-s)</li>
+ *     <li>IntegrationTestHelper (helper methods)</li>
+ * </ul>
+ */
+public class RaftClusterInvocationContext implements 
TestTemplateInvocationContext {
+
+    private final ClusterConfig clusterConfig;
+    private final AtomicReference<KafkaClusterTestKit> clusterReference;
+
+    public RaftClusterInvocationContext(ClusterConfig clusterConfig) {
+        this.clusterConfig = clusterConfig;
+        this.clusterReference = new AtomicReference<>();
+    }
+
+    @Override
+    public String getDisplayName(int invocationIndex) {
+        String clusterDesc = clusterConfig.nameTags().entrySet().stream()
+            .map(Object::toString)
+            .collect(Collectors.joining(", "));
+        return String.format("[Quorum %d] %s", invocationIndex, clusterDesc);
+    }
+
+    @Override
+    public List<Extension> getAdditionalExtensions() {
+        return Arrays.asList(
+            (BeforeTestExecutionCallback) context -> {
+                KafkaClusterTestKit.Builder builder = new 
KafkaClusterTestKit.Builder(
+                    new TestKitNodes.Builder().
+                        setNumKip500BrokerNodes(clusterConfig.numBrokers()).
+                        
setNumControllerNodes(clusterConfig.numControllers()).build());
+
+                // Copy properties into the TestKit builder
+                clusterConfig.serverProperties().forEach((key, value) -> 
builder.setConfigProp(key.toString(), value.toString()));
+                // TODO how to pass down security protocol and listener name?
+                KafkaClusterTestKit cluster = builder.build();
+                clusterReference.set(cluster);
+                cluster.format();
+                cluster.startup();
+                kafka.utils.TestUtils.waitUntilTrue(
+                    () -> cluster.kip500Brokers().get(0).currentState() == 
BrokerState.RUNNING,
+                    () -> "Broker never made it to RUNNING state.",
+                    org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS,
+                    100L);
+            },
+            (AfterTestExecutionCallback) context -> 
clusterReference.get().close(),
+            new ClusterInstanceParameterResolver(new 
RaftClusterInstance(clusterReference, clusterConfig)),
+            new GenericParameterResolver<>(clusterConfig, ClusterConfig.class)
+        );
+    }
+
+    public static class RaftClusterInstance implements ClusterInstance {
+
+        private final AtomicReference<KafkaClusterTestKit> clusterReference;
+        private final ClusterConfig clusterConfig;
+        final AtomicBoolean started = new AtomicBoolean(false);
+        final AtomicBoolean stopped = new AtomicBoolean(false);
+
+        RaftClusterInstance(AtomicReference<KafkaClusterTestKit> 
clusterReference, ClusterConfig clusterConfig) {
+            this.clusterReference = clusterReference;
+            this.clusterConfig = clusterConfig;
+        }
+
+        @Override
+        public String bootstrapServers() {
+            return 
clusterReference.get().clientProperties().getProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
+        }
+
+        @Override
+        public Collection<SocketServer> brokerSocketServers() {
+            return clusterReference.get().kip500Brokers().values().stream()
+                .map(BrokerServer::socketServer)
+                .collect(Collectors.toList());
+        }
+
+        @Override
+        public ListenerName clientListener() {
+            return ListenerName.normalised("EXTERNAL");
+        }
+
+        @Override
+        public Collection<SocketServer> controllerSocketServers() {
+            return clusterReference.get().controllers().values().stream()
+                .map(ControllerServer::socketServer)
+                .collect(Collectors.toList());
+        }
+
+        @Override
+        public SocketServer anyBrokerSocketServer() {
+            return clusterReference.get().kip500Brokers().values().stream()
+                .map(BrokerServer::socketServer)
+                .findFirst()
+                .orElseThrow(() -> new RuntimeException("No broker 
SocketServers found"));
+        }
+
+        @Override
+        public SocketServer anyControllerSocketServer() {
+            return clusterReference.get().controllers().values().stream()
+                .map(ControllerServer::socketServer)
+                .findFirst()
+                .orElseThrow(() -> new RuntimeException("No controller 
SocketServers found"));
+        }
+
+        @Override
+        public ClusterType clusterType() {
+            return ClusterType.RAFT;
+        }
+
+        @Override
+        public ClusterConfig config() {
+            return clusterConfig;
+        }
+
+        @Override
+        public Object getUnderlying() {
+            return clusterReference.get();
+        }
+
+        @Override
+        public Admin createAdminClient(Properties configOverrides) {
+            return Admin.create(clusterReference.get().clientProperties());
+        }
+
+        @Override
+        public void start() {
+            if (started.compareAndSet(false, true)) {
+                try {
+                    clusterReference.get().startup();
+                } catch (Exception e) {
+                    throw new RuntimeException("Failed to start up Raft 
server", e);
+                }
+            }
+        }
+
+        @Override
+        public void stop() {
+            if (stopped.compareAndSet(false, true)) {
+                try {
+                    clusterReference.get().close();
+                } catch (Exception e) {
+                    throw new RuntimeException("Failed to stop up Raft 
server", e);

Review comment:
       nit: drop "up"

##########
File path: core/src/test/java/kafka/test/junit/RaftClusterInvocationContext.java
##########
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.test.junit;
+
+import kafka.network.SocketServer;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.test.ClusterConfig;
+import kafka.test.ClusterInstance;
+import kafka.testkit.KafkaClusterTestKit;
+import kafka.testkit.TestKitNodes;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.metadata.BrokerState;
+import org.junit.jupiter.api.extension.AfterTestExecutionCallback;
+import org.junit.jupiter.api.extension.BeforeTestExecutionCallback;
+import org.junit.jupiter.api.extension.Extension;
+import org.junit.jupiter.api.extension.TestTemplateInvocationContext;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+/**
+ * Wraps a {@link KafkaClusterTestKit} inside lifecycle methods for a test 
invocation. Each instance of this
+ * class is provided with a configuration for the cluster.
+ *
+ * This context also provides parameter resolvers for:
+ *
+ * <ul>
+ *     <li>ClusterConfig (the same instance passed to the constructor)</li>
+ *     <li>ClusterInstance (includes methods to expose underlying 
SocketServer-s)</li>
+ *     <li>IntegrationTestHelper (helper methods)</li>
+ * </ul>
+ */
+public class RaftClusterInvocationContext implements 
TestTemplateInvocationContext {
+
+    private final ClusterConfig clusterConfig;
+    private final AtomicReference<KafkaClusterTestKit> clusterReference;
+
+    public RaftClusterInvocationContext(ClusterConfig clusterConfig) {
+        this.clusterConfig = clusterConfig;
+        this.clusterReference = new AtomicReference<>();
+    }
+
+    @Override
+    public String getDisplayName(int invocationIndex) {
+        String clusterDesc = clusterConfig.nameTags().entrySet().stream()
+            .map(Object::toString)
+            .collect(Collectors.joining(", "));
+        return String.format("[Quorum %d] %s", invocationIndex, clusterDesc);
+    }
+
+    @Override
+    public List<Extension> getAdditionalExtensions() {
+        return Arrays.asList(
+            (BeforeTestExecutionCallback) context -> {
+                KafkaClusterTestKit.Builder builder = new 
KafkaClusterTestKit.Builder(
+                    new TestKitNodes.Builder().
+                        setNumKip500BrokerNodes(clusterConfig.numBrokers()).
+                        
setNumControllerNodes(clusterConfig.numControllers()).build());
+
+                // Copy properties into the TestKit builder
+                clusterConfig.serverProperties().forEach((key, value) -> 
builder.setConfigProp(key.toString(), value.toString()));
+                // TODO how to pass down security protocol and listener name?

Review comment:
       Maybe we can turn this into a JIRA?

##########
File path: core/src/test/java/kafka/testkit/Kip500BrokerNode.java
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import org.apache.kafka.common.Uuid;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class Kip500BrokerNode implements TestKitNode {
+    public static class Builder {
+        private int id = -1;
+        private Uuid incarnationId = null;
+        private String metadataDirectory = null;

Review comment:
       It looks like the builder doesn't provide a way to set this. On a side 
note, I wonder if we should mimic the configuration behavior more closely. If 
`metadata.log.dir` is not set explicitly, then we use the first listed data 
directory.

##########
File path: core/src/test/java/kafka/testkit/TestKitNodes.java
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.server.MetaProperties;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+
+public class TestKitNodes {
+    public static class Builder {
+        private Uuid clusterId = null;
+        private final NavigableMap<Integer, ControllerNode> controllerNodes = 
new TreeMap<>();
+        private final NavigableMap<Integer, Kip500BrokerNode> 
kip500BrokerNodes = new TreeMap<>();
+
+        public Builder setClusterId(Uuid clusterId) {
+            this.clusterId = clusterId;
+            return this;
+        }
+
+        public Builder addNodes(TestKitNode[] nodes) {
+            for (TestKitNode node : nodes) {
+                addNode(node);
+            }
+            return this;
+        }
+
+        public Builder addNode(TestKitNode node) {
+            if (node instanceof ControllerNode) {
+                ControllerNode controllerNode = (ControllerNode) node;
+                controllerNodes.put(node.id(), controllerNode);
+            } else if (node instanceof Kip500BrokerNode) {
+                Kip500BrokerNode brokerNode = (Kip500BrokerNode) node;
+                kip500BrokerNodes.put(node.id(), brokerNode);
+            } else {
+                throw new RuntimeException("Can't handle TestKitNode subclass 
" +
+                        node.getClass().getSimpleName());
+            }
+            return this;
+        }
+
+        public Builder setNumControllerNodes(int numControllerNodes) {
+            if (numControllerNodes < 0) {
+                throw new RuntimeException("Invalid negative value for 
numControllerNodes");
+            }
+            while (controllerNodes.size() > numControllerNodes) {

Review comment:
       nit: this loop is a little unconventional. Maybe we could use 
`pollFirstEntry` instead of the iterator? Similarly in 
`setNumKip500BrokerNodes`. 

##########
File path: core/src/test/scala/integration/kafka/server/RaftClusterTest.scala
##########
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import kafka.testkit.{KafkaClusterTestKit, TestKitNodes}
+import kafka.utils.TestUtils
+import org.apache.kafka.clients.admin.{Admin, NewTopic}
+import org.apache.kafka.common.quota.{ClientQuotaAlteration, 
ClientQuotaEntity, ClientQuotaFilter, ClientQuotaFilterComponent}
+import org.apache.kafka.metadata.BrokerState
+import org.junit.jupiter.api.{Test, Timeout}
+import org.junit.jupiter.api.Assertions._
+
+import java.util
+import java.util.Collections
+import java.util.concurrent.TimeUnit
+import scala.jdk.CollectionConverters._
+
+@Timeout(120000)
+class RaftClusterTest {
+
+  @Test
+  def testCreateClusterAndClose(): Unit = {
+    val cluster = new KafkaClusterTestKit.Builder(
+      new TestKitNodes.Builder().
+        setNumKip500BrokerNodes(1).
+        setNumControllerNodes(1).build()).build()
+    try {
+      cluster.format()
+      cluster.startup()
+    } finally {
+      cluster.close()
+    }
+  }
+
+  @Test
+  def testCreateClusterAndWaitForBrokerInRunningState(): Unit = {
+    val cluster = new KafkaClusterTestKit.Builder(
+      new TestKitNodes.Builder().
+        setNumKip500BrokerNodes(3).
+        setNumControllerNodes(3).build()).build()
+    try {
+      cluster.format()
+      cluster.startup()
+      TestUtils.waitUntilTrue(() => 
cluster.kip500Brokers().get(0).currentState() == BrokerState.RUNNING,
+        "Broker never made it to RUNNING state.")
+      TestUtils.waitUntilTrue(() => 
cluster.raftManagers().get(0).kafkaRaftClient.leaderAndEpoch().leaderId.isPresent,
+      "RaftManager was not initialized.")
+      val admin = Admin.create(cluster.clientProperties())
+      try {
+        assertEquals(cluster.nodes().clusterId().toString,
+          admin.describeCluster().clusterId().get())
+      } finally {
+        admin.close()
+      }
+    } finally {
+      cluster.close()
+    }
+  }
+
+  @Test
+  def testCreateClusterAndCreateAndListTopic(): Unit = {
+    val cluster = new KafkaClusterTestKit.Builder(
+      new TestKitNodes.Builder().
+        setNumKip500BrokerNodes(3).
+        setNumControllerNodes(3).build()).build()
+    try {
+      cluster.format()
+      cluster.startup()
+      cluster.waitForReadyBrokers()
+      TestUtils.waitUntilTrue(() => 
cluster.kip500Brokers().get(0).currentState() == BrokerState.RUNNING,
+        "Broker never made it to RUNNING state.")
+      TestUtils.waitUntilTrue(() => 
cluster.raftManagers().get(0).kafkaRaftClient.leaderAndEpoch().leaderId.isPresent,
+        "RaftManager was not initialized.")
+      val admin = Admin.create(cluster.clientProperties())
+      try {
+        // Create a test topic
+        val newTopic = Collections.singletonList(new NewTopic("test-topic", 1, 
3.toShort))
+        val createTopicResult = admin.createTopics(newTopic)
+        createTopicResult.all().get(60, TimeUnit.SECONDS)
+
+        // List created topic
+        TestUtils.waitUntilTrue(() => {
+          val listTopicsResult = admin.listTopics()
+          val result = listTopicsResult.names().get(5, 
TimeUnit.SECONDS).size() == newTopic.size()
+          if (result) {
+            newTopic forEach(topic => {
+              assertTrue(listTopicsResult.names().get().contains(topic.name()))
+            })
+          }
+          result
+        }, "Topics created were not listed.")
+      } finally {
+        admin.close()
+      }
+    } finally {
+      cluster.close()
+    }
+  }
+
+  @Test
+  def testCreateClusterAndCreateAndManyTopics(): Unit = {

Review comment:
       Not sure there's much benefit having the single topic case covered 
separately. If it is, we can probably factor out a generalized helper. We can 
probably cover `testCreateClusterAndCreateAndManyTopicsWithManyPartitions` with 
the helper as well.

##########
File path: core/src/test/java/kafka/testkit/KafkaClusterTestKit.java
##########
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.raft.KafkaRaftManager;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.Server;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.server.MetaProperties;
+import kafka.tools.StorageTool;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.controller.Controller;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.metalog.MetaLogManager;
+import org.apache.kafka.raft.RaftConfig;
+import org.apache.kafka.raft.metadata.MetaLogRaftShim;
+import org.apache.kafka.raft.metadata.MetadataRecordSerde;
+import org.apache.kafka.test.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
+public class KafkaClusterTestKit implements AutoCloseable {
+    private final static Logger log = 
LoggerFactory.getLogger(KafkaClusterTestKit.class);
+
+    /**
+     * This class manages a future which is completed with the proper value for
+     * controller.quorum.voters once the randomly assigned ports for all the 
controllers are
+     * known.
+     */
+    private static class ControllerQuorumVotersFutureManager implements 
AutoCloseable {
+        private final int expectedControllers;
+        private final CompletableFuture<List<String>> future = new 
CompletableFuture<>();

Review comment:
       nit: I think we could use a more convenient type, such as `Map<Integer, 
InetAddressSpec>`. Ultimately this just needs to make it down to 
`KafkaNetworkChannel.updateEndpoint` so the conversion to the config value is 
unnecessary.

##########
File path: core/src/test/java/kafka/testkit/KafkaClusterTestKit.java
##########
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.raft.KafkaRaftManager;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.Server;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.server.MetaProperties;
+import kafka.tools.StorageTool;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.controller.Controller;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.metalog.MetaLogManager;
+import org.apache.kafka.raft.RaftConfig;
+import org.apache.kafka.raft.metadata.MetaLogRaftShim;
+import org.apache.kafka.raft.metadata.MetadataRecordSerde;
+import org.apache.kafka.test.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
+public class KafkaClusterTestKit implements AutoCloseable {
+    private final static Logger log = 
LoggerFactory.getLogger(KafkaClusterTestKit.class);
+
+    /**
+     * This class manages a future which is completed with the proper value for
+     * controller.quorum.voters once the randomly assigned ports for all the 
controllers are
+     * known.
+     */
+    private static class ControllerQuorumVotersFutureManager implements 
AutoCloseable {
+        private final int expectedControllers;
+        private final CompletableFuture<List<String>> future = new 
CompletableFuture<>();
+        private final Map<Integer, Integer> controllerPorts = new TreeMap<>();
+
+        ControllerQuorumVotersFutureManager(int expectedControllers) {
+            this.expectedControllers = expectedControllers;
+        }
+
+        synchronized void registerPort(int nodeId, int port) {
+            controllerPorts.put(nodeId, port);
+            if (controllerPorts.size() >= expectedControllers) {
+                future.complete(controllerPorts.entrySet().stream().
+                    map(e -> String.format("%d@localhost:%d", e.getKey(), 
e.getValue())).
+                    collect(Collectors.toList()));
+            }
+        }
+
+        void fail(Throwable e) {
+            future.completeExceptionally(e);
+        }
+
+        @Override
+        public void close() {
+            future.cancel(true);
+        }
+    }
+
+    public static class Builder {
+        private TestKitNodes nodes;
+        private Map<String, String> configProps = new HashMap<>();
+
+        public Builder(TestKitNodes nodes) {
+            this.nodes = nodes;
+        }
+
+        public Builder setConfigProp(String key, String value) {
+            this.configProps.put(key, value);
+            return this;
+        }
+
+        public KafkaClusterTestKit build() throws Exception {
+            Map<Integer, ControllerServer> controllers = new HashMap<>();
+            Map<Integer, BrokerServer> kip500Brokers = new HashMap<>();
+            Map<Integer, KafkaRaftManager> raftManagers = new HashMap<>();
+            String dummyQuorumVotersString = 
nodes.controllerNodes().keySet().stream().
+                map(controllerNode -> String.format("%d@0.0.0.0:0", 
controllerNode)).
+                collect(Collectors.joining(","));
+            /*
+              Number of threads = Total number of brokers + Total number of 
controllers + Total number of Raft Managers
+                                = Total number of brokers + Total number of 
controllers * 2
+                                  (Raft Manager per broker/controller)
+             */
+            int numOfExecutorThreads = (nodes.brokerNodes().size() + 
nodes.controllerNodes().size()) * 2;
+            ExecutorService executorService = null;
+            ControllerQuorumVotersFutureManager connectFutureManager =
+                new 
ControllerQuorumVotersFutureManager(nodes.controllerNodes().size());
+            File baseDirectory = null;
+
+            try {
+                baseDirectory = TestUtils.tempDirectory();
+                nodes = 
nodes.copyWithAbsolutePaths(baseDirectory.getAbsolutePath());
+                executorService = 
Executors.newFixedThreadPool(numOfExecutorThreads,
+                    ThreadUtils.createThreadFactory("KafkaClusterTestKit%d", 
false));
+                Time time = Time.SYSTEM;

Review comment:
       nit: this seems unnecessary. We're already using the constant below 
anyway.

##########
File path: core/src/test/java/kafka/testkit/KafkaClusterTestKit.java
##########
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.raft.KafkaRaftManager;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.Server;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.server.MetaProperties;
+import kafka.tools.StorageTool;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.controller.Controller;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.metalog.MetaLogManager;
+import org.apache.kafka.raft.RaftConfig;
+import org.apache.kafka.raft.metadata.MetaLogRaftShim;
+import org.apache.kafka.raft.metadata.MetadataRecordSerde;
+import org.apache.kafka.test.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
+public class KafkaClusterTestKit implements AutoCloseable {
+    private final static Logger log = 
LoggerFactory.getLogger(KafkaClusterTestKit.class);
+
+    /**
+     * This class manages a future which is completed with the proper value for
+     * controller.quorum.voters once the randomly assigned ports for all the 
controllers are
+     * known.
+     */
+    private static class ControllerQuorumVotersFutureManager implements 
AutoCloseable {
+        private final int expectedControllers;
+        private final CompletableFuture<List<String>> future = new 
CompletableFuture<>();
+        private final Map<Integer, Integer> controllerPorts = new TreeMap<>();
+
+        ControllerQuorumVotersFutureManager(int expectedControllers) {
+            this.expectedControllers = expectedControllers;
+        }
+
+        synchronized void registerPort(int nodeId, int port) {
+            controllerPorts.put(nodeId, port);
+            if (controllerPorts.size() >= expectedControllers) {
+                future.complete(controllerPorts.entrySet().stream().
+                    map(e -> String.format("%d@localhost:%d", e.getKey(), 
e.getValue())).
+                    collect(Collectors.toList()));
+            }
+        }
+
+        void fail(Throwable e) {
+            future.completeExceptionally(e);
+        }
+
+        @Override
+        public void close() {
+            future.cancel(true);
+        }
+    }
+
+    public static class Builder {
+        private TestKitNodes nodes;
+        private Map<String, String> configProps = new HashMap<>();
+
+        public Builder(TestKitNodes nodes) {
+            this.nodes = nodes;
+        }
+
+        public Builder setConfigProp(String key, String value) {
+            this.configProps.put(key, value);
+            return this;
+        }
+
+        public KafkaClusterTestKit build() throws Exception {
+            Map<Integer, ControllerServer> controllers = new HashMap<>();
+            Map<Integer, BrokerServer> kip500Brokers = new HashMap<>();
+            Map<Integer, KafkaRaftManager> raftManagers = new HashMap<>();
+            String dummyQuorumVotersString = 
nodes.controllerNodes().keySet().stream().
+                map(controllerNode -> String.format("%d@0.0.0.0:0", 
controllerNode)).
+                collect(Collectors.joining(","));
+            /*
+              Number of threads = Total number of brokers + Total number of 
controllers + Total number of Raft Managers
+                                = Total number of brokers + Total number of 
controllers * 2
+                                  (Raft Manager per broker/controller)
+             */
+            int numOfExecutorThreads = (nodes.brokerNodes().size() + 
nodes.controllerNodes().size()) * 2;
+            ExecutorService executorService = null;
+            ControllerQuorumVotersFutureManager connectFutureManager =
+                new 
ControllerQuorumVotersFutureManager(nodes.controllerNodes().size());
+            File baseDirectory = null;
+
+            try {
+                baseDirectory = TestUtils.tempDirectory();
+                nodes = 
nodes.copyWithAbsolutePaths(baseDirectory.getAbsolutePath());
+                executorService = 
Executors.newFixedThreadPool(numOfExecutorThreads,
+                    ThreadUtils.createThreadFactory("KafkaClusterTestKit%d", 
false));
+                Time time = Time.SYSTEM;
+                for (ControllerNode node : nodes.controllerNodes().values()) {

Review comment:
       It might be nice to factor out a helper to build the controller and 
broker nodes. It would make it a little easier to process this method visually.

##########
File path: core/src/test/java/kafka/testkit/KafkaClusterTestKit.java
##########
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.raft.KafkaRaftManager;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.Server;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.server.MetaProperties;
+import kafka.tools.StorageTool;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.controller.Controller;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.metalog.MetaLogManager;
+import org.apache.kafka.raft.RaftConfig;
+import org.apache.kafka.raft.metadata.MetaLogRaftShim;
+import org.apache.kafka.raft.metadata.MetadataRecordSerde;
+import org.apache.kafka.test.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
+public class KafkaClusterTestKit implements AutoCloseable {
+    private final static Logger log = 
LoggerFactory.getLogger(KafkaClusterTestKit.class);
+
+    /**
+     * This class manages a future which is completed with the proper value for
+     * controller.quorum.voters once the randomly assigned ports for all the 
controllers are
+     * known.
+     */
+    private static class ControllerQuorumVotersFutureManager implements 
AutoCloseable {
+        private final int expectedControllers;
+        private final CompletableFuture<List<String>> future = new 
CompletableFuture<>();
+        private final Map<Integer, Integer> controllerPorts = new TreeMap<>();
+
+        ControllerQuorumVotersFutureManager(int expectedControllers) {
+            this.expectedControllers = expectedControllers;
+        }
+
+        synchronized void registerPort(int nodeId, int port) {
+            controllerPorts.put(nodeId, port);
+            if (controllerPorts.size() >= expectedControllers) {
+                future.complete(controllerPorts.entrySet().stream().
+                    map(e -> String.format("%d@localhost:%d", e.getKey(), 
e.getValue())).
+                    collect(Collectors.toList()));
+            }
+        }
+
+        void fail(Throwable e) {
+            future.completeExceptionally(e);
+        }
+
+        @Override
+        public void close() {
+            future.cancel(true);
+        }
+    }
+
+    public static class Builder {
+        private TestKitNodes nodes;
+        private Map<String, String> configProps = new HashMap<>();
+
+        public Builder(TestKitNodes nodes) {
+            this.nodes = nodes;
+        }
+
+        public Builder setConfigProp(String key, String value) {
+            this.configProps.put(key, value);
+            return this;
+        }
+
+        public KafkaClusterTestKit build() throws Exception {
+            Map<Integer, ControllerServer> controllers = new HashMap<>();
+            Map<Integer, BrokerServer> kip500Brokers = new HashMap<>();
+            Map<Integer, KafkaRaftManager> raftManagers = new HashMap<>();
+            String dummyQuorumVotersString = 
nodes.controllerNodes().keySet().stream().
+                map(controllerNode -> String.format("%d@0.0.0.0:0", 
controllerNode)).
+                collect(Collectors.joining(","));
+            /*
+              Number of threads = Total number of brokers + Total number of 
controllers + Total number of Raft Managers
+                                = Total number of brokers + Total number of 
controllers * 2
+                                  (Raft Manager per broker/controller)
+             */
+            int numOfExecutorThreads = (nodes.brokerNodes().size() + 
nodes.controllerNodes().size()) * 2;
+            ExecutorService executorService = null;
+            ControllerQuorumVotersFutureManager connectFutureManager =
+                new 
ControllerQuorumVotersFutureManager(nodes.controllerNodes().size());
+            File baseDirectory = null;
+
+            try {
+                baseDirectory = TestUtils.tempDirectory();
+                nodes = 
nodes.copyWithAbsolutePaths(baseDirectory.getAbsolutePath());
+                executorService = 
Executors.newFixedThreadPool(numOfExecutorThreads,
+                    ThreadUtils.createThreadFactory("KafkaClusterTestKit%d", 
false));
+                Time time = Time.SYSTEM;
+                for (ControllerNode node : nodes.controllerNodes().values()) {
+                    Map<String, String> props = new HashMap<>(configProps);
+                    props.put(KafkaConfig$.MODULE$.ProcessRolesProp(), 
"controller");
+                    props.put(KafkaConfig$.MODULE$.NodeIdProp(),
+                        Integer.toString(node.id()));
+                    props.put(KafkaConfig$.MODULE$.MetadataLogDirProp(),
+                        node.metadataDirectory());
+                    
props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(),
+                        "CONTROLLER:PLAINTEXT");
+                    props.put(KafkaConfig$.MODULE$.ListenersProp(),
+                        "CONTROLLER://localhost:0");
+                    
props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(),
+                        "CONTROLLER");
+                    // Note: we can't accurately set controller.quorum.voters 
yet, since we don't
+                    // yet know what ports each controller will pick.  Set it 
to a dummy string \
+                    // for now as a placeholder.
+                    props.put(RaftConfig.QUORUM_VOTERS_CONFIG, 
dummyQuorumVotersString);

Review comment:
       nit: Instead of calling it `dummy` which makes it sound hacky, maybe we 
could call it `uninitializedQuorumVotersString` or something like that. We have 
tried to make configuring with the `0.0.0.0:0` endpoint an explicitly supported 
feature.

##########
File path: core/src/test/java/kafka/testkit/KafkaClusterTestKit.java
##########
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.raft.KafkaRaftManager;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.Server;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.server.MetaProperties;
+import kafka.tools.StorageTool;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.controller.Controller;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.metalog.MetaLogManager;
+import org.apache.kafka.raft.RaftConfig;
+import org.apache.kafka.raft.metadata.MetaLogRaftShim;
+import org.apache.kafka.raft.metadata.MetadataRecordSerde;
+import org.apache.kafka.test.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
+public class KafkaClusterTestKit implements AutoCloseable {
+    private final static Logger log = 
LoggerFactory.getLogger(KafkaClusterTestKit.class);
+
+    /**
+     * This class manages a future which is completed with the proper value for
+     * controller.quorum.voters once the randomly assigned ports for all the 
controllers are
+     * known.
+     */
+    private static class ControllerQuorumVotersFutureManager implements 
AutoCloseable {
+        private final int expectedControllers;
+        private final CompletableFuture<List<String>> future = new 
CompletableFuture<>();
+        private final Map<Integer, Integer> controllerPorts = new TreeMap<>();
+
+        ControllerQuorumVotersFutureManager(int expectedControllers) {
+            this.expectedControllers = expectedControllers;
+        }
+
+        synchronized void registerPort(int nodeId, int port) {
+            controllerPorts.put(nodeId, port);
+            if (controllerPorts.size() >= expectedControllers) {
+                future.complete(controllerPorts.entrySet().stream().
+                    map(e -> String.format("%d@localhost:%d", e.getKey(), 
e.getValue())).
+                    collect(Collectors.toList()));
+            }
+        }
+
+        void fail(Throwable e) {
+            future.completeExceptionally(e);
+        }
+
+        @Override
+        public void close() {
+            future.cancel(true);
+        }
+    }
+
+    public static class Builder {
+        private TestKitNodes nodes;
+        private Map<String, String> configProps = new HashMap<>();
+
+        public Builder(TestKitNodes nodes) {
+            this.nodes = nodes;
+        }
+
+        public Builder setConfigProp(String key, String value) {
+            this.configProps.put(key, value);
+            return this;
+        }
+
+        public KafkaClusterTestKit build() throws Exception {
+            Map<Integer, ControllerServer> controllers = new HashMap<>();
+            Map<Integer, BrokerServer> kip500Brokers = new HashMap<>();
+            Map<Integer, KafkaRaftManager> raftManagers = new HashMap<>();
+            String dummyQuorumVotersString = 
nodes.controllerNodes().keySet().stream().
+                map(controllerNode -> String.format("%d@0.0.0.0:0", 
controllerNode)).
+                collect(Collectors.joining(","));
+            /*
+              Number of threads = Total number of brokers + Total number of 
controllers + Total number of Raft Managers
+                                = Total number of brokers + Total number of 
controllers * 2
+                                  (Raft Manager per broker/controller)
+             */
+            int numOfExecutorThreads = (nodes.brokerNodes().size() + 
nodes.controllerNodes().size()) * 2;
+            ExecutorService executorService = null;
+            ControllerQuorumVotersFutureManager connectFutureManager =
+                new 
ControllerQuorumVotersFutureManager(nodes.controllerNodes().size());
+            File baseDirectory = null;
+
+            try {
+                baseDirectory = TestUtils.tempDirectory();
+                nodes = 
nodes.copyWithAbsolutePaths(baseDirectory.getAbsolutePath());
+                executorService = 
Executors.newFixedThreadPool(numOfExecutorThreads,
+                    ThreadUtils.createThreadFactory("KafkaClusterTestKit%d", 
false));
+                Time time = Time.SYSTEM;
+                for (ControllerNode node : nodes.controllerNodes().values()) {
+                    Map<String, String> props = new HashMap<>(configProps);
+                    props.put(KafkaConfig$.MODULE$.ProcessRolesProp(), 
"controller");
+                    props.put(KafkaConfig$.MODULE$.NodeIdProp(),
+                        Integer.toString(node.id()));
+                    props.put(KafkaConfig$.MODULE$.MetadataLogDirProp(),
+                        node.metadataDirectory());
+                    
props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(),
+                        "CONTROLLER:PLAINTEXT");
+                    props.put(KafkaConfig$.MODULE$.ListenersProp(),
+                        "CONTROLLER://localhost:0");
+                    
props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(),
+                        "CONTROLLER");
+                    // Note: we can't accurately set controller.quorum.voters 
yet, since we don't
+                    // yet know what ports each controller will pick.  Set it 
to a dummy string \
+                    // for now as a placeholder.
+                    props.put(RaftConfig.QUORUM_VOTERS_CONFIG, 
dummyQuorumVotersString);
+                    setupNodeDirectories(baseDirectory, 
node.metadataDirectory(), Collections.emptyList());
+                    KafkaConfig config = new KafkaConfig(props, false, 
Option.empty());
+
+                    String threadNamePrefix = String.format("controller%d_", 
node.id());
+                    MetaProperties metaProperties = 
MetaProperties.apply(nodes.clusterId(), node.id());
+                    TopicPartition metadataPartition = new 
TopicPartition("@metadata", 0);
+                    KafkaRaftManager<ApiMessageAndVersion> raftManager = new 
KafkaRaftManager<>(
+                        metaProperties, config, new MetadataRecordSerde(), 
metadataPartition,
+                        Time.SYSTEM, new Metrics(), 
Option.apply(threadNamePrefix), connectFutureManager.future);
+                    MetaLogManager metaLogShim = new 
MetaLogRaftShim(raftManager.kafkaRaftClient(), config.nodeId());
+                    ControllerServer controller = new ControllerServer(
+                        nodes.controllerProperties(node.id()),
+                        config,
+                        metaLogShim,
+                        raftManager,
+                        time,
+                        new Metrics(),
+                        Option.apply(threadNamePrefix),
+                        connectFutureManager.future
+                    );
+                    controllers.put(node.id(), controller);
+                    
controller.socketServerFirstBoundPortFuture().whenComplete((port, e) -> {
+                        if (e != null) {
+                            connectFutureManager.fail(e);
+                        } else {
+                            connectFutureManager.registerPort(node.id(), port);
+                        }
+                    });
+                    raftManagers.put(node.id(), raftManager);
+                }
+                for (Kip500BrokerNode node : nodes.brokerNodes().values()) {

Review comment:
       By the way, I sort of feel it would make our lives easier if we used 
`KafkaRaftServer` directly instead of building the controller, broker, and raft 
managers ourselves. For one thing, that would make it trivial to support mixed 
mode. We don't have to do that here, but I'm kind of curious if there is a 
reason that we don't.

##########
File path: core/src/test/java/kafka/testkit/KafkaClusterTestKit.java
##########
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.raft.KafkaRaftManager;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.Server;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.server.MetaProperties;
+import kafka.tools.StorageTool;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.controller.Controller;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.metalog.MetaLogManager;
+import org.apache.kafka.raft.RaftConfig;
+import org.apache.kafka.raft.metadata.MetaLogRaftShim;
+import org.apache.kafka.raft.metadata.MetadataRecordSerde;
+import org.apache.kafka.test.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
+public class KafkaClusterTestKit implements AutoCloseable {
+    private final static Logger log = 
LoggerFactory.getLogger(KafkaClusterTestKit.class);
+
+    /**
+     * This class manages a future which is completed with the proper value for
+     * controller.quorum.voters once the randomly assigned ports for all the 
controllers are
+     * known.
+     */
+    private static class ControllerQuorumVotersFutureManager implements 
AutoCloseable {
+        private final int expectedControllers;
+        private final CompletableFuture<List<String>> future = new 
CompletableFuture<>();
+        private final Map<Integer, Integer> controllerPorts = new TreeMap<>();
+
+        ControllerQuorumVotersFutureManager(int expectedControllers) {
+            this.expectedControllers = expectedControllers;
+        }
+
+        synchronized void registerPort(int nodeId, int port) {
+            controllerPorts.put(nodeId, port);
+            if (controllerPorts.size() >= expectedControllers) {
+                future.complete(controllerPorts.entrySet().stream().
+                    map(e -> String.format("%d@localhost:%d", e.getKey(), 
e.getValue())).
+                    collect(Collectors.toList()));
+            }
+        }
+
+        void fail(Throwable e) {
+            future.completeExceptionally(e);
+        }
+
+        @Override
+        public void close() {
+            future.cancel(true);
+        }
+    }
+
+    public static class Builder {
+        private TestKitNodes nodes;
+        private Map<String, String> configProps = new HashMap<>();
+
+        public Builder(TestKitNodes nodes) {
+            this.nodes = nodes;
+        }
+
+        public Builder setConfigProp(String key, String value) {
+            this.configProps.put(key, value);
+            return this;
+        }
+
+        public KafkaClusterTestKit build() throws Exception {
+            Map<Integer, ControllerServer> controllers = new HashMap<>();
+            Map<Integer, BrokerServer> kip500Brokers = new HashMap<>();
+            Map<Integer, KafkaRaftManager> raftManagers = new HashMap<>();
+            String dummyQuorumVotersString = 
nodes.controllerNodes().keySet().stream().
+                map(controllerNode -> String.format("%d@0.0.0.0:0", 
controllerNode)).
+                collect(Collectors.joining(","));
+            /*
+              Number of threads = Total number of brokers + Total number of 
controllers + Total number of Raft Managers
+                                = Total number of brokers + Total number of 
controllers * 2
+                                  (Raft Manager per broker/controller)
+             */
+            int numOfExecutorThreads = (nodes.brokerNodes().size() + 
nodes.controllerNodes().size()) * 2;
+            ExecutorService executorService = null;
+            ControllerQuorumVotersFutureManager connectFutureManager =
+                new 
ControllerQuorumVotersFutureManager(nodes.controllerNodes().size());
+            File baseDirectory = null;
+
+            try {
+                baseDirectory = TestUtils.tempDirectory();
+                nodes = 
nodes.copyWithAbsolutePaths(baseDirectory.getAbsolutePath());
+                executorService = 
Executors.newFixedThreadPool(numOfExecutorThreads,
+                    ThreadUtils.createThreadFactory("KafkaClusterTestKit%d", 
false));
+                Time time = Time.SYSTEM;
+                for (ControllerNode node : nodes.controllerNodes().values()) {
+                    Map<String, String> props = new HashMap<>(configProps);
+                    props.put(KafkaConfig$.MODULE$.ProcessRolesProp(), 
"controller");
+                    props.put(KafkaConfig$.MODULE$.NodeIdProp(),
+                        Integer.toString(node.id()));
+                    props.put(KafkaConfig$.MODULE$.MetadataLogDirProp(),
+                        node.metadataDirectory());
+                    
props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(),
+                        "CONTROLLER:PLAINTEXT");
+                    props.put(KafkaConfig$.MODULE$.ListenersProp(),
+                        "CONTROLLER://localhost:0");
+                    
props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(),
+                        "CONTROLLER");
+                    // Note: we can't accurately set controller.quorum.voters 
yet, since we don't
+                    // yet know what ports each controller will pick.  Set it 
to a dummy string \
+                    // for now as a placeholder.
+                    props.put(RaftConfig.QUORUM_VOTERS_CONFIG, 
dummyQuorumVotersString);
+                    setupNodeDirectories(baseDirectory, 
node.metadataDirectory(), Collections.emptyList());
+                    KafkaConfig config = new KafkaConfig(props, false, 
Option.empty());
+
+                    String threadNamePrefix = String.format("controller%d_", 
node.id());
+                    MetaProperties metaProperties = 
MetaProperties.apply(nodes.clusterId(), node.id());
+                    TopicPartition metadataPartition = new 
TopicPartition("@metadata", 0);
+                    KafkaRaftManager<ApiMessageAndVersion> raftManager = new 
KafkaRaftManager<>(
+                        metaProperties, config, new MetadataRecordSerde(), 
metadataPartition,
+                        Time.SYSTEM, new Metrics(), 
Option.apply(threadNamePrefix), connectFutureManager.future);
+                    MetaLogManager metaLogShim = new 
MetaLogRaftShim(raftManager.kafkaRaftClient(), config.nodeId());
+                    ControllerServer controller = new ControllerServer(
+                        nodes.controllerProperties(node.id()),
+                        config,
+                        metaLogShim,
+                        raftManager,
+                        time,
+                        new Metrics(),
+                        Option.apply(threadNamePrefix),
+                        connectFutureManager.future
+                    );
+                    controllers.put(node.id(), controller);
+                    
controller.socketServerFirstBoundPortFuture().whenComplete((port, e) -> {
+                        if (e != null) {
+                            connectFutureManager.fail(e);
+                        } else {
+                            connectFutureManager.registerPort(node.id(), port);
+                        }
+                    });
+                    raftManagers.put(node.id(), raftManager);
+                }
+                for (Kip500BrokerNode node : nodes.brokerNodes().values()) {
+                    Map<String, String> props = new HashMap<>(configProps);
+                    props.put(KafkaConfig$.MODULE$.ProcessRolesProp(), 
"broker");
+                    props.put(KafkaConfig$.MODULE$.BrokerIdProp(),
+                        Integer.toString(node.id()));
+                    props.put(KafkaConfig$.MODULE$.MetadataLogDirProp(),
+                        node.metadataDirectory());
+                    props.put(KafkaConfig$.MODULE$.LogDirsProp(),
+                        String.join(",", node.logDataDirectories()));
+                    
props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(),
+                        "EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT");
+                    props.put(KafkaConfig$.MODULE$.ListenersProp(),
+                        "EXTERNAL://localhost:0");
+                    
props.put(KafkaConfig$.MODULE$.InterBrokerListenerNameProp(),
+                        nodes.interBrokerListenerName().value());
+                    
props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(),
+                        "CONTROLLER");
+
+                    setupNodeDirectories(baseDirectory, 
node.metadataDirectory(),
+                        node.logDataDirectories());
+
+                    // Just like above, we set a placeholder voter list here 
until we
+                    // find out what ports the controllers picked.
+                    props.put(RaftConfig.QUORUM_VOTERS_CONFIG, 
dummyQuorumVotersString);
+                    KafkaConfig config = new KafkaConfig(props, false, 
Option.empty());
+
+                    String threadNamePrefix = String.format("broker%d_", 
node.id());
+                    MetaProperties metaProperties = 
MetaProperties.apply(nodes.clusterId(), node.id());
+                    TopicPartition metadataPartition = new 
TopicPartition("@metadata", 0);
+                    KafkaRaftManager<ApiMessageAndVersion> raftManager = new 
KafkaRaftManager<>(
+                            metaProperties, config, new MetadataRecordSerde(), 
metadataPartition,
+                            Time.SYSTEM, new Metrics(), 
Option.apply(threadNamePrefix), connectFutureManager.future);
+                    MetaLogManager metaLogShim = new 
MetaLogRaftShim(raftManager.kafkaRaftClient(), config.nodeId());
+                    BrokerServer broker = new BrokerServer(
+                        config,
+                        nodes.brokerProperties(node.id()),
+                        metaLogShim,
+                        time,
+                        new Metrics(),
+                        Option.apply(threadNamePrefix),
+                        
JavaConverters.asScalaBuffer(Collections.<String>emptyList()).toSeq(),
+                        connectFutureManager.future,
+                        Server.SUPPORTED_FEATURES()
+                    );
+                    kip500Brokers.put(node.id(), broker);
+                    raftManagers.put(node.id(), raftManager);
+                }
+            } catch (Exception e) {
+                if (executorService != null) {
+                    executorService.shutdownNow();
+                    executorService.awaitTermination(1, TimeUnit.DAYS);

Review comment:
       Would we really wait one day for tests to shutdown? Maybe one minute is 
enough?

##########
File path: core/src/test/java/kafka/testkit/KafkaClusterTestKit.java
##########
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.raft.KafkaRaftManager;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.Server;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.server.MetaProperties;
+import kafka.tools.StorageTool;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.controller.Controller;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.metalog.MetaLogManager;
+import org.apache.kafka.raft.RaftConfig;
+import org.apache.kafka.raft.metadata.MetaLogRaftShim;
+import org.apache.kafka.raft.metadata.MetadataRecordSerde;
+import org.apache.kafka.test.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
+public class KafkaClusterTestKit implements AutoCloseable {
+    private final static Logger log = 
LoggerFactory.getLogger(KafkaClusterTestKit.class);
+
+    /**
+     * This class manages a future which is completed with the proper value for
+     * controller.quorum.voters once the randomly assigned ports for all the 
controllers are
+     * known.
+     */
+    private static class ControllerQuorumVotersFutureManager implements 
AutoCloseable {
+        private final int expectedControllers;
+        private final CompletableFuture<List<String>> future = new 
CompletableFuture<>();
+        private final Map<Integer, Integer> controllerPorts = new TreeMap<>();
+
+        ControllerQuorumVotersFutureManager(int expectedControllers) {
+            this.expectedControllers = expectedControllers;
+        }
+
+        synchronized void registerPort(int nodeId, int port) {
+            controllerPorts.put(nodeId, port);
+            if (controllerPorts.size() >= expectedControllers) {
+                future.complete(controllerPorts.entrySet().stream().
+                    map(e -> String.format("%d@localhost:%d", e.getKey(), 
e.getValue())).
+                    collect(Collectors.toList()));
+            }
+        }
+
+        void fail(Throwable e) {
+            future.completeExceptionally(e);
+        }
+
+        @Override
+        public void close() {
+            future.cancel(true);
+        }
+    }
+
+    public static class Builder {
+        private TestKitNodes nodes;
+        private Map<String, String> configProps = new HashMap<>();
+
+        public Builder(TestKitNodes nodes) {
+            this.nodes = nodes;
+        }
+
+        public Builder setConfigProp(String key, String value) {
+            this.configProps.put(key, value);
+            return this;
+        }
+
+        public KafkaClusterTestKit build() throws Exception {
+            Map<Integer, ControllerServer> controllers = new HashMap<>();
+            Map<Integer, BrokerServer> kip500Brokers = new HashMap<>();
+            Map<Integer, KafkaRaftManager> raftManagers = new HashMap<>();
+            String dummyQuorumVotersString = 
nodes.controllerNodes().keySet().stream().
+                map(controllerNode -> String.format("%d@0.0.0.0:0", 
controllerNode)).
+                collect(Collectors.joining(","));
+            /*
+              Number of threads = Total number of brokers + Total number of 
controllers + Total number of Raft Managers
+                                = Total number of brokers + Total number of 
controllers * 2
+                                  (Raft Manager per broker/controller)
+             */
+            int numOfExecutorThreads = (nodes.brokerNodes().size() + 
nodes.controllerNodes().size()) * 2;
+            ExecutorService executorService = null;
+            ControllerQuorumVotersFutureManager connectFutureManager =
+                new 
ControllerQuorumVotersFutureManager(nodes.controllerNodes().size());
+            File baseDirectory = null;
+
+            try {
+                baseDirectory = TestUtils.tempDirectory();
+                nodes = 
nodes.copyWithAbsolutePaths(baseDirectory.getAbsolutePath());
+                executorService = 
Executors.newFixedThreadPool(numOfExecutorThreads,
+                    ThreadUtils.createThreadFactory("KafkaClusterTestKit%d", 
false));
+                Time time = Time.SYSTEM;
+                for (ControllerNode node : nodes.controllerNodes().values()) {
+                    Map<String, String> props = new HashMap<>(configProps);
+                    props.put(KafkaConfig$.MODULE$.ProcessRolesProp(), 
"controller");
+                    props.put(KafkaConfig$.MODULE$.NodeIdProp(),
+                        Integer.toString(node.id()));
+                    props.put(KafkaConfig$.MODULE$.MetadataLogDirProp(),
+                        node.metadataDirectory());
+                    
props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(),
+                        "CONTROLLER:PLAINTEXT");
+                    props.put(KafkaConfig$.MODULE$.ListenersProp(),
+                        "CONTROLLER://localhost:0");
+                    
props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(),
+                        "CONTROLLER");
+                    // Note: we can't accurately set controller.quorum.voters 
yet, since we don't
+                    // yet know what ports each controller will pick.  Set it 
to a dummy string \
+                    // for now as a placeholder.
+                    props.put(RaftConfig.QUORUM_VOTERS_CONFIG, 
dummyQuorumVotersString);
+                    setupNodeDirectories(baseDirectory, 
node.metadataDirectory(), Collections.emptyList());
+                    KafkaConfig config = new KafkaConfig(props, false, 
Option.empty());
+
+                    String threadNamePrefix = String.format("controller%d_", 
node.id());
+                    MetaProperties metaProperties = 
MetaProperties.apply(nodes.clusterId(), node.id());
+                    TopicPartition metadataPartition = new 
TopicPartition("@metadata", 0);
+                    KafkaRaftManager<ApiMessageAndVersion> raftManager = new 
KafkaRaftManager<>(
+                        metaProperties, config, new MetadataRecordSerde(), 
metadataPartition,
+                        Time.SYSTEM, new Metrics(), 
Option.apply(threadNamePrefix), connectFutureManager.future);
+                    MetaLogManager metaLogShim = new 
MetaLogRaftShim(raftManager.kafkaRaftClient(), config.nodeId());
+                    ControllerServer controller = new ControllerServer(
+                        nodes.controllerProperties(node.id()),
+                        config,
+                        metaLogShim,
+                        raftManager,
+                        time,
+                        new Metrics(),
+                        Option.apply(threadNamePrefix),
+                        connectFutureManager.future
+                    );
+                    controllers.put(node.id(), controller);
+                    
controller.socketServerFirstBoundPortFuture().whenComplete((port, e) -> {
+                        if (e != null) {
+                            connectFutureManager.fail(e);
+                        } else {
+                            connectFutureManager.registerPort(node.id(), port);
+                        }
+                    });
+                    raftManagers.put(node.id(), raftManager);
+                }
+                for (Kip500BrokerNode node : nodes.brokerNodes().values()) {
+                    Map<String, String> props = new HashMap<>(configProps);
+                    props.put(KafkaConfig$.MODULE$.ProcessRolesProp(), 
"broker");
+                    props.put(KafkaConfig$.MODULE$.BrokerIdProp(),
+                        Integer.toString(node.id()));
+                    props.put(KafkaConfig$.MODULE$.MetadataLogDirProp(),
+                        node.metadataDirectory());
+                    props.put(KafkaConfig$.MODULE$.LogDirsProp(),
+                        String.join(",", node.logDataDirectories()));
+                    
props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(),
+                        "EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT");
+                    props.put(KafkaConfig$.MODULE$.ListenersProp(),
+                        "EXTERNAL://localhost:0");
+                    
props.put(KafkaConfig$.MODULE$.InterBrokerListenerNameProp(),
+                        nodes.interBrokerListenerName().value());
+                    
props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(),
+                        "CONTROLLER");
+
+                    setupNodeDirectories(baseDirectory, 
node.metadataDirectory(),
+                        node.logDataDirectories());
+
+                    // Just like above, we set a placeholder voter list here 
until we
+                    // find out what ports the controllers picked.
+                    props.put(RaftConfig.QUORUM_VOTERS_CONFIG, 
dummyQuorumVotersString);
+                    KafkaConfig config = new KafkaConfig(props, false, 
Option.empty());
+
+                    String threadNamePrefix = String.format("broker%d_", 
node.id());
+                    MetaProperties metaProperties = 
MetaProperties.apply(nodes.clusterId(), node.id());
+                    TopicPartition metadataPartition = new 
TopicPartition("@metadata", 0);
+                    KafkaRaftManager<ApiMessageAndVersion> raftManager = new 
KafkaRaftManager<>(
+                            metaProperties, config, new MetadataRecordSerde(), 
metadataPartition,
+                            Time.SYSTEM, new Metrics(), 
Option.apply(threadNamePrefix), connectFutureManager.future);
+                    MetaLogManager metaLogShim = new 
MetaLogRaftShim(raftManager.kafkaRaftClient(), config.nodeId());
+                    BrokerServer broker = new BrokerServer(
+                        config,
+                        nodes.brokerProperties(node.id()),
+                        metaLogShim,
+                        time,
+                        new Metrics(),
+                        Option.apply(threadNamePrefix),
+                        
JavaConverters.asScalaBuffer(Collections.<String>emptyList()).toSeq(),
+                        connectFutureManager.future,
+                        Server.SUPPORTED_FEATURES()
+                    );
+                    kip500Brokers.put(node.id(), broker);
+                    raftManagers.put(node.id(), raftManager);
+                }
+            } catch (Exception e) {
+                if (executorService != null) {
+                    executorService.shutdownNow();
+                    executorService.awaitTermination(1, TimeUnit.DAYS);
+                }
+                for (ControllerServer controller : controllers.values()) {
+                    controller.shutdown();
+                }
+                for (BrokerServer brokerServer : kip500Brokers.values()) {
+                    brokerServer.shutdown();
+                }
+                for (KafkaRaftManager raftManager : raftManagers.values()) {
+                    raftManager.shutdown();
+                }
+                connectFutureManager.close();
+                if (baseDirectory != null) {
+                    Utils.delete(baseDirectory);
+                }
+                throw e;
+            }
+            return new KafkaClusterTestKit(executorService, nodes, controllers,
+                kip500Brokers, raftManagers, connectFutureManager, 
baseDirectory);
+        }
+
+        static private void setupNodeDirectories(File baseDirectory,
+                                                 String metadataDirectory,
+                                                 Collection<String> 
logDataDirectories) throws Exception {
+            Files.createDirectories(new File(baseDirectory, "local").toPath());
+            Files.createDirectories(Paths.get(metadataDirectory));
+            for (String logDataDirectory : logDataDirectories) {
+                Files.createDirectories(Paths.get(logDataDirectory));
+            }
+        }
+    }
+
+    private final ExecutorService executorService;
+    private final TestKitNodes nodes;
+    private final Map<Integer, ControllerServer> controllers;
+    private final Map<Integer, BrokerServer> kip500Brokers;
+    private final Map<Integer, KafkaRaftManager> raftManagers;
+    private final ControllerQuorumVotersFutureManager 
controllerQuorumVotersFutureManager;
+    private final File baseDirectory;
+
+    private KafkaClusterTestKit(ExecutorService executorService,
+                                TestKitNodes nodes,
+                                Map<Integer, ControllerServer> controllers,
+                                Map<Integer, BrokerServer> kip500Brokers,
+                                Map<Integer, KafkaRaftManager> raftManagers,
+                                ControllerQuorumVotersFutureManager 
controllerQuorumVotersFutureManager,
+                                File baseDirectory) {
+        this.executorService = executorService;
+        this.nodes = nodes;
+        this.controllers = controllers;
+        this.kip500Brokers = kip500Brokers;
+        this.raftManagers = raftManagers;
+        this.controllerQuorumVotersFutureManager = 
controllerQuorumVotersFutureManager;
+        this.baseDirectory = baseDirectory;
+    }
+
+    public void format() throws Exception {
+        List<Future<?>> futures = new ArrayList<>();
+        try {
+            for (Entry<Integer, ControllerServer> entry : 
controllers.entrySet()) {
+                int nodeId = entry.getKey();

Review comment:
       We really ought to be able to factor out a helper here. The code looks 
identical for both the controller and broker cases.

##########
File path: core/src/test/java/kafka/testkit/KafkaClusterTestKit.java
##########
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.raft.KafkaRaftManager;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.Server;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.server.MetaProperties;
+import kafka.tools.StorageTool;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.controller.Controller;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.metalog.MetaLogManager;
+import org.apache.kafka.raft.RaftConfig;
+import org.apache.kafka.raft.metadata.MetaLogRaftShim;
+import org.apache.kafka.raft.metadata.MetadataRecordSerde;
+import org.apache.kafka.test.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
+public class KafkaClusterTestKit implements AutoCloseable {
+    private final static Logger log = 
LoggerFactory.getLogger(KafkaClusterTestKit.class);
+
+    /**
+     * This class manages a future which is completed with the proper value for
+     * controller.quorum.voters once the randomly assigned ports for all the 
controllers are
+     * known.
+     */
+    private static class ControllerQuorumVotersFutureManager implements 
AutoCloseable {
+        private final int expectedControllers;
+        private final CompletableFuture<List<String>> future = new 
CompletableFuture<>();
+        private final Map<Integer, Integer> controllerPorts = new TreeMap<>();
+
+        ControllerQuorumVotersFutureManager(int expectedControllers) {
+            this.expectedControllers = expectedControllers;
+        }
+
+        synchronized void registerPort(int nodeId, int port) {
+            controllerPorts.put(nodeId, port);
+            if (controllerPorts.size() >= expectedControllers) {
+                future.complete(controllerPorts.entrySet().stream().
+                    map(e -> String.format("%d@localhost:%d", e.getKey(), 
e.getValue())).
+                    collect(Collectors.toList()));
+            }
+        }
+
+        void fail(Throwable e) {
+            future.completeExceptionally(e);
+        }
+
+        @Override
+        public void close() {
+            future.cancel(true);
+        }
+    }
+
+    public static class Builder {
+        private TestKitNodes nodes;
+        private Map<String, String> configProps = new HashMap<>();
+
+        public Builder(TestKitNodes nodes) {
+            this.nodes = nodes;
+        }
+
+        public Builder setConfigProp(String key, String value) {
+            this.configProps.put(key, value);
+            return this;
+        }
+
+        public KafkaClusterTestKit build() throws Exception {
+            Map<Integer, ControllerServer> controllers = new HashMap<>();
+            Map<Integer, BrokerServer> kip500Brokers = new HashMap<>();
+            Map<Integer, KafkaRaftManager> raftManagers = new HashMap<>();
+            String dummyQuorumVotersString = 
nodes.controllerNodes().keySet().stream().
+                map(controllerNode -> String.format("%d@0.0.0.0:0", 
controllerNode)).
+                collect(Collectors.joining(","));
+            /*
+              Number of threads = Total number of brokers + Total number of 
controllers + Total number of Raft Managers
+                                = Total number of brokers + Total number of 
controllers * 2
+                                  (Raft Manager per broker/controller)
+             */
+            int numOfExecutorThreads = (nodes.brokerNodes().size() + 
nodes.controllerNodes().size()) * 2;
+            ExecutorService executorService = null;
+            ControllerQuorumVotersFutureManager connectFutureManager =
+                new 
ControllerQuorumVotersFutureManager(nodes.controllerNodes().size());
+            File baseDirectory = null;
+
+            try {
+                baseDirectory = TestUtils.tempDirectory();
+                nodes = 
nodes.copyWithAbsolutePaths(baseDirectory.getAbsolutePath());
+                executorService = 
Executors.newFixedThreadPool(numOfExecutorThreads,
+                    ThreadUtils.createThreadFactory("KafkaClusterTestKit%d", 
false));
+                Time time = Time.SYSTEM;
+                for (ControllerNode node : nodes.controllerNodes().values()) {
+                    Map<String, String> props = new HashMap<>(configProps);
+                    props.put(KafkaConfig$.MODULE$.ProcessRolesProp(), 
"controller");
+                    props.put(KafkaConfig$.MODULE$.NodeIdProp(),
+                        Integer.toString(node.id()));
+                    props.put(KafkaConfig$.MODULE$.MetadataLogDirProp(),
+                        node.metadataDirectory());
+                    
props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(),
+                        "CONTROLLER:PLAINTEXT");
+                    props.put(KafkaConfig$.MODULE$.ListenersProp(),
+                        "CONTROLLER://localhost:0");
+                    
props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(),
+                        "CONTROLLER");
+                    // Note: we can't accurately set controller.quorum.voters 
yet, since we don't
+                    // yet know what ports each controller will pick.  Set it 
to a dummy string \
+                    // for now as a placeholder.
+                    props.put(RaftConfig.QUORUM_VOTERS_CONFIG, 
dummyQuorumVotersString);
+                    setupNodeDirectories(baseDirectory, 
node.metadataDirectory(), Collections.emptyList());
+                    KafkaConfig config = new KafkaConfig(props, false, 
Option.empty());
+
+                    String threadNamePrefix = String.format("controller%d_", 
node.id());
+                    MetaProperties metaProperties = 
MetaProperties.apply(nodes.clusterId(), node.id());
+                    TopicPartition metadataPartition = new 
TopicPartition("@metadata", 0);

Review comment:
       I don't suppose we could reference the constant defined in 
`KafkaRaftServer`?

##########
File path: core/src/test/scala/integration/kafka/server/RaftClusterTest.scala
##########
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import kafka.testkit.{KafkaClusterTestKit, TestKitNodes}
+import kafka.utils.TestUtils
+import org.apache.kafka.clients.admin.{Admin, NewTopic}
+import org.apache.kafka.common.quota.{ClientQuotaAlteration, 
ClientQuotaEntity, ClientQuotaFilter, ClientQuotaFilterComponent}
+import org.apache.kafka.metadata.BrokerState
+import org.junit.jupiter.api.{Test, Timeout}
+import org.junit.jupiter.api.Assertions._
+
+import java.util
+import java.util.Collections
+import java.util.concurrent.TimeUnit
+import scala.jdk.CollectionConverters._
+
+@Timeout(120000)
+class RaftClusterTest {
+
+  @Test
+  def testCreateClusterAndClose(): Unit = {
+    val cluster = new KafkaClusterTestKit.Builder(
+      new TestKitNodes.Builder().
+        setNumKip500BrokerNodes(1).
+        setNumControllerNodes(1).build()).build()
+    try {
+      cluster.format()
+      cluster.startup()
+    } finally {
+      cluster.close()
+    }
+  }
+
+  @Test
+  def testCreateClusterAndWaitForBrokerInRunningState(): Unit = {
+    val cluster = new KafkaClusterTestKit.Builder(
+      new TestKitNodes.Builder().
+        setNumKip500BrokerNodes(3).
+        setNumControllerNodes(3).build()).build()
+    try {
+      cluster.format()
+      cluster.startup()
+      TestUtils.waitUntilTrue(() => 
cluster.kip500Brokers().get(0).currentState() == BrokerState.RUNNING,
+        "Broker never made it to RUNNING state.")
+      TestUtils.waitUntilTrue(() => 
cluster.raftManagers().get(0).kafkaRaftClient.leaderAndEpoch().leaderId.isPresent,
+      "RaftManager was not initialized.")

Review comment:
       nit: alignment looks off

##########
File path: core/src/test/resources/log4j.properties
##########
@@ -18,8 +18,8 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n
 
-log4j.logger.kafka=ERROR
-log4j.logger.org.apache.kafka=ERROR
+log4j.logger.kafka=INFO

Review comment:
       Guessing this was just added for debugging.

##########
File path: core/src/test/java/kafka/testkit/TestKitNodes.java
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.server.MetaProperties;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.network.ListenerName;
+
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+
+public class TestKitNodes {
+    public static class Builder {
+        private Uuid clusterId = null;
+        private final NavigableMap<Integer, ControllerNode> controllerNodes = 
new TreeMap<>();

Review comment:
       It looks like we don't support mixed mode testing. That seems worth a 
follow-up JIRA. It is definitely an interesting case from the perspective of 
the raft implementation since it involves two listeners.

##########
File path: core/src/test/scala/integration/kafka/server/RaftClusterTest.scala
##########
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+import kafka.testkit.{KafkaClusterTestKit, TestKitNodes}
+import kafka.utils.TestUtils
+import org.apache.kafka.clients.admin.{Admin, NewTopic}
+import org.apache.kafka.common.quota.{ClientQuotaAlteration, 
ClientQuotaEntity, ClientQuotaFilter, ClientQuotaFilterComponent}
+import org.apache.kafka.metadata.BrokerState
+import org.junit.jupiter.api.{Test, Timeout}
+import org.junit.jupiter.api.Assertions._
+
+import java.util
+import java.util.Collections
+import java.util.concurrent.TimeUnit
+import scala.jdk.CollectionConverters._
+
+@Timeout(120000)
+class RaftClusterTest {
+
+  @Test
+  def testCreateClusterAndClose(): Unit = {
+    val cluster = new KafkaClusterTestKit.Builder(
+      new TestKitNodes.Builder().
+        setNumKip500BrokerNodes(1).
+        setNumControllerNodes(1).build()).build()
+    try {
+      cluster.format()
+      cluster.startup()
+    } finally {
+      cluster.close()
+    }
+  }
+
+  @Test
+  def testCreateClusterAndWaitForBrokerInRunningState(): Unit = {
+    val cluster = new KafkaClusterTestKit.Builder(
+      new TestKitNodes.Builder().
+        setNumKip500BrokerNodes(3).
+        setNumControllerNodes(3).build()).build()
+    try {
+      cluster.format()
+      cluster.startup()
+      TestUtils.waitUntilTrue(() => 
cluster.kip500Brokers().get(0).currentState() == BrokerState.RUNNING,
+        "Broker never made it to RUNNING state.")
+      TestUtils.waitUntilTrue(() => 
cluster.raftManagers().get(0).kafkaRaftClient.leaderAndEpoch().leaderId.isPresent,
+      "RaftManager was not initialized.")
+      val admin = Admin.create(cluster.clientProperties())
+      try {
+        assertEquals(cluster.nodes().clusterId().toString,
+          admin.describeCluster().clusterId().get())
+      } finally {
+        admin.close()
+      }
+    } finally {
+      cluster.close()
+    }
+  }
+
+  @Test
+  def testCreateClusterAndCreateAndListTopic(): Unit = {

Review comment:
       Maybe we can let this cover topic deletion as well

##########
File path: core/src/test/java/kafka/testkit/KafkaClusterTestKit.java
##########
@@ -0,0 +1,499 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.testkit;
+
+import kafka.raft.KafkaRaftManager;
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaConfig$;
+import kafka.server.Server;
+import kafka.server.BrokerServer;
+import kafka.server.ControllerServer;
+import kafka.server.MetaProperties;
+import kafka.tools.StorageTool;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.ThreadUtils;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.controller.Controller;
+import org.apache.kafka.metadata.ApiMessageAndVersion;
+import org.apache.kafka.metalog.MetaLogManager;
+import org.apache.kafka.raft.RaftConfig;
+import org.apache.kafka.raft.metadata.MetaLogRaftShim;
+import org.apache.kafka.raft.metadata.MetadataRecordSerde;
+import org.apache.kafka.test.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+
+@SuppressWarnings("deprecation") // Needed for Scala 2.12 compatibility
+public class KafkaClusterTestKit implements AutoCloseable {
+    private final static Logger log = 
LoggerFactory.getLogger(KafkaClusterTestKit.class);
+
+    /**
+     * This class manages a future which is completed with the proper value for
+     * controller.quorum.voters once the randomly assigned ports for all the 
controllers are
+     * known.
+     */
+    private static class ControllerQuorumVotersFutureManager implements 
AutoCloseable {
+        private final int expectedControllers;
+        private final CompletableFuture<List<String>> future = new 
CompletableFuture<>();
+        private final Map<Integer, Integer> controllerPorts = new TreeMap<>();
+
+        ControllerQuorumVotersFutureManager(int expectedControllers) {
+            this.expectedControllers = expectedControllers;
+        }
+
+        synchronized void registerPort(int nodeId, int port) {
+            controllerPorts.put(nodeId, port);
+            if (controllerPorts.size() >= expectedControllers) {
+                future.complete(controllerPorts.entrySet().stream().
+                    map(e -> String.format("%d@localhost:%d", e.getKey(), 
e.getValue())).
+                    collect(Collectors.toList()));
+            }
+        }
+
+        void fail(Throwable e) {
+            future.completeExceptionally(e);
+        }
+
+        @Override
+        public void close() {
+            future.cancel(true);
+        }
+    }
+
+    public static class Builder {
+        private TestKitNodes nodes;
+        private Map<String, String> configProps = new HashMap<>();
+
+        public Builder(TestKitNodes nodes) {
+            this.nodes = nodes;
+        }
+
+        public Builder setConfigProp(String key, String value) {
+            this.configProps.put(key, value);
+            return this;
+        }
+
+        public KafkaClusterTestKit build() throws Exception {
+            Map<Integer, ControllerServer> controllers = new HashMap<>();
+            Map<Integer, BrokerServer> kip500Brokers = new HashMap<>();
+            Map<Integer, KafkaRaftManager> raftManagers = new HashMap<>();
+            String dummyQuorumVotersString = 
nodes.controllerNodes().keySet().stream().

Review comment:
       Just want to point out that this assumes all controllers are voters. It 
would be worth a follow-up to support controllers as observers as well.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org