Do not use ownerReference for EO role in separate watched namespace

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

@@ -384,12 +384,15 @@ public static String getRoleName(String cluster) {
     /**
      * Read the entity operator ClusterRole, and use the rules to create a new Role.
      * This is done to avoid duplication of the rules set defined in source code.
+     * If the namespace of the role is not the same as the namespace of the parent resource (Kafka CR), we do not set
+     * the owner reference.
      *
-     * @param namespace the namespace this role will be located
+     * @param ownerNamespace        The namespace of the parent resource (the Kafka CR)
+     * @param namespace             The namespace this role will be located
      *
      * @return role for the entity operator
      */
-    public Role generateRole(String namespace) {
+    public Role generateRole(String ownerNamespace, String namespace) {
         List<PolicyRule> rules;
 
         try (BufferedReader br = new BufferedReader(
@@ -407,7 +410,14 @@ public Role generateRole(String namespace) {
             throw new RuntimeException(e);
         }
 
-        return super.generateRole(namespace, rules);
+        Role role = super.generateRole(namespace, rules);
+
+        // We set OwnerReference only within the same namespace since it does not work cross-namespace
+        if (!namespace.equals(ownerNamespace)) {
+            role.getMetadata().setOwnerReferences(Collections.emptyList());
+        }
+
+        return role;
     }
 
     protected static void javaOptions(List<EnvVar> envVars, JvmOptions jvmOptions, List<SystemProperty> javaSystemProperties) {

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaAssemblyOperator.java

@@ -162,8 +162,6 @@ public class KafkaAssemblyOperator extends AbstractAssemblyOperator<KubernetesCl
     private final String operatorNamespace;
     private final Labels operatorNamespaceLabels;
 
-    private final ClusterOperatorConfig.RbacScope rbacScope;
-
     private final ZookeeperSetOperator zkSetOperations;
     private final KafkaSetOperator kafkaSetOperations;
     private final RouteOperator routeOperations;
@@ -197,7 +195,6 @@ public KafkaAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa,
         this.operationTimeoutMs = config.getOperationTimeoutMs();
         this.operatorNamespace = config.getOperatorNamespace();
         this.operatorNamespaceLabels = config.getOperatorNamespaceLabels();
-        this.rbacScope = config.getRbacScope();
         this.routeOperations = supplier.routeOperations;
         this.zkSetOperations = supplier.zkSetOperations;
         this.kafkaSetOperations = supplier.kafkaSetOperations;
@@ -3057,7 +3054,7 @@ final Future<ReconciliationState> getEntityOperatorDescription() {
         Future<ReconciliationState> entityOperatorRole() {
             final Role role;
             if (isEntityOperatorDeployed()) {
-                role = entityOperator.generateRole(namespace);
+                role = entityOperator.generateRole(namespace, namespace);
             } else {
                 role = null;
             }
@@ -3085,7 +3082,7 @@ Future<ReconciliationState> entityTopicOperatorRole() {
                 topicWatchedNamespaceFuture = roleOperations.reconcile(
                         topicWatchedNamespace,
                         EntityOperator.getRoleName(name),
-                        entityOperator.generateRole(topicWatchedNamespace));
+                        entityOperator.generateRole(namespace, topicWatchedNamespace));
             } else {
                 topicWatchedNamespaceFuture = Future.succeededFuture();
             }
@@ -3110,7 +3107,7 @@ Future<ReconciliationState> entityUserOperatorRole() {
                 userWatchedNamespaceFuture = roleOperations.reconcile(
                         userWatchedNamespace,
                         EntityOperator.getRoleName(name),
-                        entityOperator.generateRole(userWatchedNamespace));
+                        entityOperator.generateRole(namespace, userWatchedNamespace));
             } else {
                 userWatchedNamespaceFuture = Future.succeededFuture();
             }

cluster-operator/src/test/java/io/strimzi/operator/cluster/model/EntityOperatorTest.java

@@ -895,7 +895,6 @@ public void testTlsSidecarContainerSecurityContext() {
 
     @Test
     public void testRole() {
-
         Kafka resource = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout))
                 .editSpec()
                     .editOrNewEntityOperator()
@@ -904,7 +903,7 @@ public void testRole() {
                 .build();
 
         EntityOperator eo =  EntityOperator.fromCrd(resource, VERSIONS);
-        Role role = eo.generateRole(namespace);
+        Role role = eo.generateRole(namespace, namespace);
 
         assertThat(role.getMetadata().getName(), is("foo-entity-operator"));
         assertThat(role.getMetadata().getNamespace(), is(namespace));
@@ -927,4 +926,22 @@ public void testRole() {
                 .build());
         assertThat(role.getRules(), is(rules));
     }
+
+    @Test
+    public void testRoleInDifferentNamespace() {
+        Kafka resource = new KafkaBuilder(ResourceUtils.createKafka(namespace, cluster, replicas, image, healthDelay, healthTimeout))
+                .editSpec()
+                .editOrNewEntityOperator()
+                .endEntityOperator()
+                .endSpec()
+                .build();
+
+        EntityOperator eo =  EntityOperator.fromCrd(resource, VERSIONS);
+        Role role = eo.generateRole(namespace, namespace);
+
+        assertThat(role.getMetadata().getOwnerReferences().get(0), is(entityOperator.createOwnerReference()));
+
+        role = eo.generateRole(namespace, "some-other-namespace");
+        assertThat(role.getMetadata().getOwnerReferences().size(), is(0));
+    }
 }