Address comments in Operator

Signed-off-by: ocorriga <ocorriga@redhat.com>

Author: ocorriga

operator/src/main/java/io/strimzi/kafka/access/KafkaAccessOperator.java

@@ -6,21 +6,49 @@
 
 import io.javaoperatorsdk.operator.Operator;
 import io.strimzi.kafka.access.server.HealthServlet;
-import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.Set;
-import java.util.stream.Collectors;
+import org.apache.kafka.common.errors.InvalidConfigurationException;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.servlet.ServletHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static java.util.Arrays.asList;
+
 /**
  * The main operator class for Strimzi Access Operator
  */
 public class KafkaAccessOperator {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(KafkaAccessOperator.class);
     private static final int HEALTH_CHECK_PORT = 8080;
+    private final static String ANY_NAMESPACE = "*";
+
+    /**
+     * Parses namespace configuration from a string
+     *
+     * @param namespacesList    Comma-separated list of namespaces or "*" for all namespaces
+     * @return                  Set of namespace names
+     */
+    private static Set<String> parseNamespaces(String namespacesList) {
+        Set<String> namespaces;
+        if (namespacesList.equals(ANY_NAMESPACE)) {
+            namespaces = Collections.singleton(ANY_NAMESPACE);
+        } else {
+            if (namespacesList.trim().equals(ANY_NAMESPACE)) {
+                namespaces = Collections.singleton(ANY_NAMESPACE);
+            } else if (namespacesList.matches("(\\s*[a-z0-9.-]+\\s*,)*\\s*[a-z0-9.-]+\\s*")) {
+                namespaces = new HashSet<>(asList(namespacesList.trim().split("\\s*,+\\s*")));
+            } else {
+                throw new InvalidConfigurationException("Not a valid list of namespaces nor the 'any namespace' wildcard "
+                        + ANY_NAMESPACE);
+            }
+        }
+
+        return namespaces;
+    }
 
     /**
      * Initializes the operator and runs a servlet for health checking
@@ -32,22 +60,23 @@ public static void main(final String[] args) {
         final Operator operator = new Operator(overrider -> overrider
                 .withUseSSAToPatchPrimaryResource(false));
 
-        String strimziNamespace = System.getenv("STRIMZI_NAMESPACE");
-        if (strimziNamespace != null && strimziNamespace.matches("\\*")) {
-            LOGGER.info("Watching all namespaces");
-            operator.register(new KafkaAccessReconciler(operator.getKubernetesClient()),
-                    o -> o.watchingAllNamespaces());
-        } else {
-            Set<String> namespaces =
-                    strimziNamespace == null ? Set.of() :
-                            Arrays.stream(strimziNamespace.split(","))
-                                    .map(String::trim)
-                                    .collect(Collectors.toSet());
-            LOGGER.info("Watching specific namespaces: {}", namespaces);
-            operator.register(new KafkaAccessReconciler(operator.getKubernetesClient()),
-                    o -> o.settingNamespaces(namespaces));
+        String strimziNamespace = System.getenv().getOrDefault("STRIMZI_NAMESPACE", "*");
+        Set<String> namespaces = parseNamespaces(strimziNamespace);
+
+        operator.register(new KafkaAccessReconciler(operator.getKubernetesClient()),
+            configuration -> configuration.settingNamespaces(namespaces));
+
+        try {
+            operator.start();
+            LOGGER.info("Kafka Access operator started successfully for namespaces: {}", namespaces);
+        } catch (Exception e) {
+            LOGGER.error("Failed to start Kafka Access operator. This may be due to missing RoleBindings for one or more namespaces: {}. " +
+                    "Please ensure that appropriate RoleBindings exist for all configured namespaces. Error: {}",
+                    namespaces, e.getMessage(), e);
+            // Continue running to keep pod alive and allow health checks to work
+            LOGGER.warn("Operator will continue running but may not be able to reconcile resources in namespaces without proper permissions");
         }
-        operator.start();
+
         Server server = new Server(HEALTH_CHECK_PORT);
         ServletHandler handler = new ServletHandler();
         server.setHandler(handler);

operator/src/test/java/io/strimzi/kafka/access/KafkaAccessReconcilerTest.java

@@ -560,23 +560,16 @@ void testReconcileWithUserProvidedSecretAndTestDeleteSecretWithNameChange() {
         assertThat(oldSecretBeforeRename).isNotNull();
         assertThat(oldSecretBeforeRename.getType()).isEqualTo("servicebinding.io/kafka");
 
-        // Retry update to handle race condition with reconciler status updates
-        boolean updated = false;
-        for (int i = 0; i < 5 && !updated; i++) {
-            try {
-                KafkaAccess currentKafkaAccess = client.resources(KafkaAccess.class).inNamespace(NAMESPACE).withName(NAME).get();
-                assertThat(currentKafkaAccess).isNotNull();
-                currentKafkaAccess.getSpec().setSecretName(NEW_USER_PROVIDED_SECRET_NAME);
-                client.resources(KafkaAccess.class).resource(currentKafkaAccess).update();
-                updated = true;
-            } catch (io.fabric8.kubernetes.client.KubernetesClientException e) {
-                if (e.getCode() != 409) {
-                    throw e;
-                }
-                // Retry on 409 conflict
-            }
-        }
-        assertThat(updated).as("Failed to update KafkaAccess after retries").isTrue();
+        KafkaAccess currentKafkaAccess = client.resources(KafkaAccess.class).inNamespace(NAMESPACE).withName(NAME).get();
+        assertThat(currentKafkaAccess).isNotNull();
+
+        client.resources(KafkaAccess.class)
+                .inNamespace(NAMESPACE)
+                .withName(NAME)
+                .edit(ka -> {
+                    ka.getSpec().setSecretName(NEW_USER_PROVIDED_SECRET_NAME);
+                    return ka;
+                });
 
         client.resources(KafkaAccess.class).inNamespace(NAMESPACE).withName(NAME).waitUntilCondition(updatedKafkaAccess -> {
             final Optional<String> bindingName = Optional.ofNullable(updatedKafkaAccess)

packaging/install/030-ClusterRoleBinding.yaml

@@ -8,7 +8,7 @@ metadata:
 subjects:
   - kind: ServiceAccount
     name: strimzi-access-operator
-    namespace: strimzi-access-operator
+    namespace: myproject
 roleRef:
   kind: ClusterRole
   name: strimzi-access-operator