Allow KafkaRoller connect to controllers (#11838)

Signed-off-by: Gantigmaa Selenge <tina.selenge@gmail.com>

Author: tinaselenge

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaCluster.java

@@ -155,7 +155,10 @@ public class KafkaCluster extends AbstractModel implements SupportsMetrics, Supp
     protected static final String REPLICATION_PORT_NAME = "tcp-replication";
     protected static final int KAFKA_AGENT_PORT = 8443;
     protected static final String KAFKA_AGENT_PORT_NAME = "tcp-kafkaagent";
-    protected static final int CONTROLPLANE_PORT = 9090;
+    /**
+     * Port number used for control plane
+     */
+    public static final int CONTROLPLANE_PORT = 9090;
     protected static final String CONTROLPLANE_PORT_NAME = "tcp-ctrlplane"; // port name is up to 15 characters
 
     /**

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/resource/KafkaRoller.java

@@ -34,9 +34,7 @@
 import org.apache.kafka.clients.admin.AlterConfigOp;
 import org.apache.kafka.clients.admin.AlterConfigsResult;
 import org.apache.kafka.clients.admin.Config;
-import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.KafkaFuture;
-import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.errors.SslAuthenticationException;
 
@@ -190,7 +188,7 @@ public KafkaRoller(Reconciliation reconciliation, Vertx vertx, PodOperator podOp
     private boolean maybeInitBrokerAdminClient() {
         if (this.brokerAdminClient == null) {
             try {
-                this.brokerAdminClient = adminClient(nodes.stream().filter(NodeRef::broker).collect(Collectors.toSet()), false);
+                this.brokerAdminClient = adminClient(nodes, false);
             } catch (ForceableProblem | FatalProblem e) {
                 LOGGER.warnCr(reconciliation, "Failed to create brokerAdminClient.", e);
                 return false;
@@ -206,13 +204,7 @@ private boolean maybeInitBrokerAdminClient() {
     private boolean maybeInitControllerAdminClient() {
         if (this.controllerAdminClient == null) {
             try {
-                // TODO: Currently, when running in KRaft mode Kafka does not support using Kafka Admin API with controller
-                //       nodes. This is tracked in https://github.com/strimzi/strimzi-kafka-operator/issues/9692.
-                //       Therefore use broker nodes of the cluster to initialise adminClient for quorum health check.
-                //       Once Kafka Admin API is supported for controllers, nodes.stream().filter(NodeRef:controller)
-                //       can be used here. Until then pass an empty set of nodes so the client is initialized with
-                //       the brokers service.
-                this.controllerAdminClient = adminClient(Set.of(), false);
+                this.controllerAdminClient = adminClient(nodes, true);
             } catch (ForceableProblem | FatalProblem e) {
                 LOGGER.warnCr(reconciliation, "Failed to create controllerAdminClient.", e);
                 return false;
@@ -400,7 +392,7 @@ private void restartIfNecessary(NodeRef nodeRef, RestartContext restartContext)
         if (!restartContext.podStuck) {
             // We want to give pods chance to get ready before we try to connect to the or consider them for rolling.
             // This is important especially for pods which were just started. But only in case when they are not stuck.
-            // If the pod is stuck, it suggests that it is running already for some time and it will not become ready.
+            // If the pod is stuck, it suggests that it is running already for some time, and it will not become ready.
             // Waiting for it would likely just waste time.
             LOGGER.debugCr(reconciliation, "Waiting for pod {} to become ready before checking its state", nodeRef.podName());
             try {
@@ -583,32 +575,12 @@ private void checkIfRestartOrReconfigureRequired(NodeRef nodeRef, boolean isCont
 
         if (isController) {
             if (maybeInitControllerAdminClient()) {
-                String controllerQuorumFetchTimeout = CONTROLLER_QUORUM_FETCH_TIMEOUT_MS_CONFIG_DEFAULT;
-                String desiredConfig = kafkaConfigProvider.apply(nodeRef.nodeId());
-
-                if (desiredConfig != null) {
-                    OrderedProperties orderedProperties = new OrderedProperties();
-                    controllerQuorumFetchTimeout = orderedProperties.addStringPairs(desiredConfig).asMap().getOrDefault(CONTROLLER_QUORUM_FETCH_TIMEOUT_MS_CONFIG_NAME, CONTROLLER_QUORUM_FETCH_TIMEOUT_MS_CONFIG_DEFAULT);
-                }
-
-                restartContext.quorumCheck = quorumCheck(controllerAdminClient, Long.parseLong(controllerQuorumFetchTimeout));
+                restartContext.quorumCheck = quorumCheck(controllerAdminClient, nodeRef);
             } else {
-                //TODO When https://github.com/strimzi/strimzi-kafka-operator/issues/9692 is complete
-                // we should change this logic to immediately restart this pod because we cannot connect to it.
-                if (isBroker) {
-                    // If it is a combined node (controller and broker) and the admin client cannot be initialised,
-                    // restart this pod. There is no reason to continue as we won't be able to
-                    // connect an admin client to this pod for other checks later.
-                    LOGGER.infoCr(reconciliation, "KafkaQuorumCheck cannot be initialised for {} because none of the brokers do not seem to responding to connection attempts. " +
-                            "Restarting pod because it is a combined node so it is one of the brokers that is not responding.", nodeRef);
-                    reasonToRestartPod.add(RestartReason.POD_UNRESPONSIVE);
-                    markRestartContextWithForceRestart(restartContext);
-                    return;
-                } else {
-                    // If it is a controller only node throw an UnforceableProblem, so we try again until the backOff
-                    // is finished, then it will move on to the next controller and eventually the brokers.
-                    throw new UnforceableProblem("KafkaQuorumCheck cannot be initialised for " + nodeRef + " because none of the brokers do not seem to responding to connection attempts");
-                }
+                LOGGER.infoCr(reconciliation, "Pod {} needs to be restarted, because it does not seem to responding to connection attempts", nodeRef);
+                reasonToRestartPod.add(RestartReason.POD_UNRESPONSIVE);
+                markRestartContextWithForceRestart(restartContext);
+                return;
             }
         }
 
@@ -811,7 +783,6 @@ private void awaitReadiness(Pod pod, long timeout, TimeUnit unit) throws FatalPr
      * @param <E> The exception type
      * @return The result of the future
      * @throws E The exception type returned from {@code exceptionMapper}.
-     * @throws TimeoutException If the given future is not completed before the timeout.
      * @throws InterruptedException If the waiting was interrupted.
      */
     private static <T, E extends Exception> T await(Future<T> future, long timeout, TimeUnit unit,
@@ -855,36 +826,35 @@ protected Future<Void> restart(Pod pod, RestartContext restartContext) {
      * Returns an AdminClient instance bootstrapped from the given nodes. If nodes is an
      * empty set, use the brokers service to bootstrap the client.
      */
-    /* test */ Admin adminClient(Set<NodeRef> nodes, boolean ceShouldBeFatal) throws ForceableProblem, FatalProblem {
-        // If no nodes are passed initialize the admin client using the brokers service
-        // TODO when https://github.com/strimzi/strimzi-kafka-operator/issues/9692 is completed review whether
-        //      this function can be reverted to expect nodes to be non empty
-        String bootstrapHostnames;
-        if (nodes.isEmpty()) {
-            bootstrapHostnames = String.format("%s:%s", DnsNameGenerator.of(namespace, KafkaResources.bootstrapServiceName(cluster)).serviceDnsName(), KafkaCluster.REPLICATION_PORT);
-        } else {
-            bootstrapHostnames = nodes.stream().map(node -> DnsNameGenerator.podDnsName(namespace, KafkaResources.brokersServiceName(cluster), node.podName()) + ":" + KafkaCluster.REPLICATION_PORT).collect(Collectors.joining(","));
-        }
-
+    /* test */ Admin adminClient(Set<NodeRef> nodes, boolean isController) throws ForceableProblem, FatalProblem {
+        String bootstrapHostnames = null;
         try {
-            LOGGER.debugCr(reconciliation, "Creating AdminClient for {}", bootstrapHostnames);
-            return adminClientProvider.createAdminClient(bootstrapHostnames, coTlsPemIdentity.pemTrustSet(), coTlsPemIdentity.pemAuthIdentity());
-        } catch (KafkaException e) {
-            if (ceShouldBeFatal && (e instanceof ConfigException
-                    || e.getCause() instanceof ConfigException)) {
-                throw new FatalProblem("An error while try to create an admin client with bootstrap brokers " + bootstrapHostnames, e);
+            if (isController) {
+                bootstrapHostnames = nodes.stream().filter(NodeRef::controller).map(node -> DnsNameGenerator.podDnsName(namespace, KafkaResources.brokersServiceName(cluster), node.podName()) + ":" + KafkaCluster.CONTROLPLANE_PORT).collect(Collectors.joining(","));
+                LOGGER.debugCr(reconciliation, "Creating AdminClient for {}", bootstrapHostnames);
+                return adminClientProvider.createControllerAdminClient(bootstrapHostnames, coTlsPemIdentity.pemTrustSet(), coTlsPemIdentity.pemAuthIdentity());
             } else {
-                throw new ForceableProblem("An error while try to create an admin client with bootstrap brokers " + bootstrapHostnames, e);
+                bootstrapHostnames = nodes.stream().filter(NodeRef::broker).map(node -> DnsNameGenerator.podDnsName(namespace, KafkaResources.brokersServiceName(cluster), node.podName()) + ":" + KafkaCluster.REPLICATION_PORT).collect(Collectors.joining(","));
+                LOGGER.debugCr(reconciliation, "Creating AdminClient for {}", bootstrapHostnames);
+                return adminClientProvider.createAdminClient(bootstrapHostnames, coTlsPemIdentity.pemTrustSet(), coTlsPemIdentity.pemAuthIdentity());
             }
         } catch (RuntimeException e) {
-            throw new ForceableProblem("An error while try to create an admin client with bootstrap brokers " + bootstrapHostnames, e);
+            throw new ForceableProblem("An error while try to create an admin client with bootstrap " + bootstrapHostnames, e);
         }
     }
 
-    /* test */ KafkaQuorumCheck quorumCheck(Admin ac, long controllerQuorumFetchTimeoutMs) {
-        return new KafkaQuorumCheck(reconciliation, ac, vertx, controllerQuorumFetchTimeoutMs);
+    /* test */ KafkaQuorumCheck quorumCheck(Admin ac, NodeRef nodeRef) {
+        String controllerQuorumFetchTimeout = CONTROLLER_QUORUM_FETCH_TIMEOUT_MS_CONFIG_DEFAULT;
+        String desiredConfig = kafkaConfigProvider.apply(nodeRef.nodeId());
+
+        if (desiredConfig != null) {
+            OrderedProperties orderedProperties = new OrderedProperties();
+            controllerQuorumFetchTimeout = orderedProperties.addStringPairs(desiredConfig).asMap().getOrDefault(CONTROLLER_QUORUM_FETCH_TIMEOUT_MS_CONFIG_NAME, CONTROLLER_QUORUM_FETCH_TIMEOUT_MS_CONFIG_DEFAULT);
+        }
+        return new KafkaQuorumCheck(reconciliation, ac, vertx, Long.parseLong(controllerQuorumFetchTimeout));
     }
 
+
     /* test */ KafkaAvailability availability(Admin ac) {
         return new KafkaAvailability(reconciliation, ac);
     }

cluster-operator/src/test/java/io/strimzi/operator/cluster/ResourceUtils.java

@@ -376,10 +376,20 @@ public Admin createAdminClient(String bootstrapHostnames, PemTrustSet kafkaCaTru
                 return createAdminClient(bootstrapHostnames, kafkaCaTrustSet, authIdentity, new Properties());
             }
 
+            @Override
+            public Admin createControllerAdminClient(String controllerBootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity) {
+                return createControllerAdminClient(controllerBootstrapHostnames, kafkaCaTrustSet, authIdentity, new Properties());
+            }
+
             @Override
             public Admin createAdminClient(String bootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config) {
                 return mockAdminClient;
             }
+
+            @Override
+            public Admin createControllerAdminClient(String controllerBootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config) {
+                return mockAdminClient;
+            }
         };
     }
 

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaRollerTest.java

@@ -997,7 +997,7 @@ Future<Boolean> canRoll(int podId) {
         }
 
         @Override
-        protected KafkaQuorumCheck quorumCheck(Admin ac, long controllerQuorumFetchTimeoutMs) {
+        protected KafkaQuorumCheck quorumCheck(Admin ac, NodeRef ref) {
             Admin admin = mock(Admin.class);
             DescribeMetadataQuorumResult qrmResult = mock(DescribeMetadataQuorumResult.class);
             when(admin.describeMetadataQuorum()).thenReturn(qrmResult);

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/events/KubernetesRestartEventsMockTest.java

@@ -588,10 +588,20 @@ public Admin createAdminClient(String bootstrapHostnames, PemTrustSet kafkaCaTru
                 return adminClientSupplier.get();
             }
 
+            @Override
+            public Admin createControllerAdminClient(String controllerBootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity) {
+                return adminClientSupplier.get();
+            }
+
             @Override
             public Admin createAdminClient(String bootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config) {
                 return adminClientSupplier.get();
             }
+
+            @Override
+            public Admin createControllerAdminClient(String controllerBootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config) {
+                return adminClientSupplier.get();
+            }
         };
 
         return new ResourceOperatorSupplier(vertx,

operator-common/src/main/java/io/strimzi/operator/common/AdminClientProvider.java

@@ -16,7 +16,7 @@
 public interface AdminClientProvider {
 
     /**
-     * Create a Kafka Admin interface instance
+     * Create a Kafka Admin interface instance for brokers
      *
      * @param bootstrapHostnames Kafka hostname to connect to for administration operations
      * @param kafkaCaTrustSet Trust set for connecting to Kafka
@@ -26,7 +26,17 @@ public interface AdminClientProvider {
     Admin createAdminClient(String bootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity);
 
     /**
-     * Create a Kafka Admin interface instance
+     * Create a Kafka Admin interface instance for controllers
+     *
+     * @param controllerBootstrapHostnames Kafka controller hostname to connect to for administration operations
+     * @param kafkaCaTrustSet Trust set for connecting to Kafka
+     * @param authIdentity Identity for TLS client authentication for connecting to Kafka
+     * @return Instance of Kafka Admin interface
+     */
+    Admin createControllerAdminClient(String controllerBootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity);
+
+    /**
+     * Create a Kafka Admin interface instance for brokers
      *
      * @param bootstrapHostnames Kafka hostname to connect to for administration operations
      * @param kafkaCaTrustSet Trust set for connecting to Kafka
@@ -36,4 +46,16 @@ public interface AdminClientProvider {
      * @return Instance of Kafka Admin interface
      */
     Admin createAdminClient(String bootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config);
+
+    /**
+     * Create a Kafka Admin interface instance for controllers
+     *
+     * @param controllerBootstrapHostnames Kafka hostname to connect to for administration operations
+     * @param kafkaCaTrustSet Trust set for connecting to Kafka
+     * @param authIdentity Identity for TLS client authentication for connecting to Kafka
+     * @param config Additional configuration for the Kafka Admin Client
+     *
+     * @return Instance of Kafka Admin interface
+     */
+    Admin createControllerAdminClient(String controllerBootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config);
 }

operator-common/src/main/java/io/strimzi/operator/common/DefaultAdminClientProvider.java

@@ -21,6 +21,11 @@ public Admin createAdminClient(String bootstrapHostnames, PemTrustSet kafkaCaTru
         return createAdminClient(bootstrapHostnames, kafkaCaTrustSet, authIdentity, new Properties());
     }
 
+    @Override
+    public Admin createControllerAdminClient(String controllerBootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity) {
+        return createControllerAdminClient(controllerBootstrapHostnames, kafkaCaTrustSet, authIdentity, new Properties());
+    }
+
     /**
      * Create a Kafka Admin interface instance handling the following different scenarios:
      *
@@ -44,26 +49,30 @@ public Admin createAdminClient(String bootstrapHostnames, PemTrustSet kafkaCaTru
      */
     @Override
     public Admin createAdminClient(String bootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config) {
-        return Admin.create(adminClientConfiguration(bootstrapHostnames, kafkaCaTrustSet, authIdentity, config));
+        config.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapHostnames);
+        return Admin.create(adminClientConfiguration(kafkaCaTrustSet, authIdentity, config));
+    }
+
+    @Override
+    public Admin createControllerAdminClient(String controllerBootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config) {
+        config.setProperty(AdminClientConfig.BOOTSTRAP_CONTROLLERS_CONFIG, controllerBootstrapHostnames);
+        return Admin.create(adminClientConfiguration(kafkaCaTrustSet, authIdentity, config));
     }
 
     /**
      * Utility method for preparing the Admin client configuration
      *
-     * @param bootstrapHostnames    Kafka bootstrap address
      * @param kafkaCaTrustSet       Trust set for connecting to Kafka
      * @param authIdentity          Identity for TLS client authentication for connecting to Kafka
      * @param config                Custom Admin client configuration or empty properties instance
      *
      * @return  Admin client configuration
      */
-    /* test */ static Properties adminClientConfiguration(String bootstrapHostnames, PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config)    {
+    /* test */ Properties adminClientConfiguration(PemTrustSet kafkaCaTrustSet, PemAuthIdentity authIdentity, Properties config)    {
         if (config == null) {
             throw new InvalidConfigurationException("The config parameter should not be null");
         }
 
-        config.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapHostnames);
-
         // configuring TLS encryption if requested
         if (kafkaCaTrustSet != null) {
             config.putIfAbsent(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SSL");