Add end to end tests itesting ability to form client connections to controller(s) (IN_VM only at the moment).

Signed-off-by: kwall <[email protected]>

Author: k-wall

api/src/main/java/io/kroxylicious/testing/kafka/api/KafkaCluster.java

@@ -105,6 +105,13 @@ public interface KafkaCluster extends AutoCloseable {
      */
     String getBootstrapServers();
 
+    /**
+     * Gets the bootstrap controllers for this cluster
+     * @return bootstrap controllers
+     * @throws UnsupportedOperationException zookeeper based clusters do not support this operation.
+     */
+    String getBootstrapControllers() throws UnsupportedOperationException;
+
     /**
      * Gets the cluster id
      * @return The cluster id for KRaft-based clusters, otherwise null;
@@ -132,4 +139,16 @@ public interface KafkaCluster extends AutoCloseable {
      * @return mutable configuration map
      */
     Map<String, Object> getKafkaClientConfiguration(String user, String password);
+
+    /**
+     * Gets the kafka configuration for making connections to the controllers of this cluster as required by the
+     * {@code org.apache.kafka.clients.admin.AdminClient}. Details such the bootstrap and SASL configuration
+     * are provided automatically.
+     * The returned map is guaranteed to be mutable and is unique to the caller.
+     *
+     * @return mutable configuration map
+     * @throws UnsupportedOperationException zookeeper based clusters do not support this operation.
+     */
+    Map<String, Object> getControllerAdminClientConfiguration() throws UnsupportedOperationException;
+
 }

impl/src/main/java/io/kroxylicious/testing/kafka/common/KafkaClusterConfig.java

@@ -619,6 +619,12 @@ public Map<String, Object> getConnectConfigForCluster(String bootstrapServers, S
         return kafkaConfig;
     }
 
+    public Map<String, Object> getControllerAdminClientConfigForCluster(String bootstrapControllers) {
+        var kafkaConfig = new HashMap<String, Object>();
+        kafkaConfig.put("bootstrap.controllers", bootstrapControllers);
+        return kafkaConfig;
+    }
+
     private void buildSecurityProtocolConfig(Map<String, Object> kafkaConfig) {
         String clientTrustStoreFilePath;
         String clientTrustStorePassword;

impl/src/main/java/io/kroxylicious/testing/kafka/invm/InVMKafkaCluster.java

@@ -260,6 +260,15 @@ public synchronized String getBootstrapServers() {
         return buildBrokerList(nodeId -> getEndpointPair(Listener.EXTERNAL, nodeId));
     }
 
+    @Override
+    public String getBootstrapControllers() {
+        if (!clusterConfig.isKraftMode()) {
+            throw new UnsupportedOperationException("Zookeeper based clusters don't support this operation");
+        }
+
+        return buildBrokerList(nodeId -> getEndpointPair(Listener.CONTROLLER, nodeId));
+    }
+
     private synchronized String buildBrokerList(Function<Integer, KafkaClusterConfig.KafkaEndpoints.EndpointPair> endpointFunc) {
         return servers.keySet().stream()
                 .filter(this::isBroker)
@@ -278,6 +287,11 @@ public Map<String, Object> getKafkaClientConfiguration(String user, String passw
         return clusterConfig.getConnectConfigForCluster(getBootstrapServers(), user, password);
     }
 
+    @Override
+    public Map<String, Object> getControllerAdminClientConfiguration() {
+        return clusterConfig.getControllerAdminClientConfigForCluster(getBootstrapControllers());
+    }
+
     @Override
     public synchronized int addBroker() {
         // find next free kafka node.id

impl/src/main/java/io/kroxylicious/testing/kafka/testcontainers/TestcontainersKafkaCluster.java

@@ -274,6 +274,11 @@ public synchronized String getBootstrapServers() {
         return buildBrokerList(nodeId -> this.getEndpointPair(Listener.EXTERNAL, nodeId));
     }
 
+    @Override
+    public String getBootstrapControllers() {
+        throw new UnsupportedOperationException();
+    }
+
     private synchronized String buildBrokerList(Function<Integer, KafkaClusterConfig.KafkaEndpoints.EndpointPair> endpointFunc) {
         return nodes.keySet().stream()
                 .filter(this::isBroker)
@@ -590,6 +595,11 @@ public Map<String, Object> getKafkaClientConfiguration(String user, String passw
         return clusterConfig.getConnectConfigForCluster(getBootstrapServers(), user, password);
     }
 
+    @Override
+    public Map<String, Object> getControllerAdminClientConfiguration() {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public synchronized EndpointPair getEndpointPair(Listener listener, int nodeId) {
         switch (listener) {

impl/src/test/java/io/kroxylicious/testing/kafka/KafkaClusterTest.java

@@ -51,12 +51,14 @@
 import io.kroxylicious.testing.kafka.common.KafkaClusterFactory;
 import io.kroxylicious.testing.kafka.common.KeytoolCertificateGenerator;
 import io.kroxylicious.testing.kafka.common.Utils;
+import io.kroxylicious.testing.kafka.invm.InVMKafkaCluster;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 /**
  * Test case that simply exercises the ability to control the kafka cluster from the test.
@@ -491,6 +493,51 @@ void kraftClusterWithMinBootstrapInterBrokerProtocol() throws Exception {
         }
     }
 
+    /**
+     * KIP-919 tests ability to connect to the controller.bootstrap.
+     */
+    @ParameterizedTest
+    @ValueSource(ints = { 1, 2 })
+    void kraftAdminConnectionToControllers(int numControllers) throws Exception {
+        int brokersNum = 1;
+        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
+                .testInfo(testInfo)
+                .brokersNum(brokersNum)
+                .kraftControllers(numControllers)
+                .kraftMode(true)
+                .build())) {
+            assumeTrue(cluster instanceof InVMKafkaCluster, "admin client connections to KRaft controllers not yet supported.");
+            cluster.start();
+
+            try (var controllerAdmin = CloseableAdmin.create(cluster.getControllerAdminClientConfiguration())) {
+                var nodes = controllerAdmin.describeCluster().nodes().get(5, TimeUnit.SECONDS);
+                assertThat(nodes).hasSize(numControllers);
+            }
+
+            try (var brokerAdmin = CloseableAdmin.create(cluster.getKafkaClientConfiguration())) {
+                var nodes = brokerAdmin.describeCluster().nodes().get(5, TimeUnit.SECONDS);
+                assertThat(nodes).hasSize(brokersNum);
+            }
+        }
+    }
+
+    @Test
+    void zookeeperDisallowsAdminConnectionToControllers() throws Exception {
+        try (var cluster = KafkaClusterFactory.create(KafkaClusterConfig.builder()
+                .testInfo(testInfo)
+                .kraftMode(false)
+                .build())) {
+            cluster.start();
+
+            assertThatThrownBy(cluster::getBootstrapControllers)
+                    .isInstanceOf(UnsupportedOperationException.class);
+
+            assertThatThrownBy(cluster::getControllerAdminClientConfiguration)
+                    .isInstanceOf(UnsupportedOperationException.class);
+        }
+
+    }
+
     private void verifyRecordRoundTrip(int expected, KafkaCluster cluster) throws Exception {
         var topic = "roundTrip" + Uuid.randomUuid();
         var message = "Hello, world!";