Fix all operators to be cluster wide

pom.xml

@@ -465,7 +465,7 @@
                                     <executable>sh</executable>
                                     <arguments>
                                         <argument>-c</argument>
-                                        <argument>helm template flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator > ${basedir}/${operator.files.destination}/flink/deploy.yaml</argument>
+                                        <argument>helm template flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator --set watchNamespaces=""  --set podSecurityContext=null --namespace flink-kubernetes-operator > ${basedir}/${operator.files.destination}/flink/deploy.yaml</argument>
                                     </arguments>
                                 </configuration>
                             </execution>

src/main/java/io/streams/constants/TestTags.java

@@ -6,5 +6,6 @@
 
 public interface TestTags {
     String SMOKE = "smoke";
+    String SQL_EXAMPLE = "sql-example";
     String DUMMY = "dummy";
 }

src/main/java/io/streams/operators/manifests/ApicurioRegistryManifestInstaller.java

@@ -4,10 +4,14 @@
  */
 package io.streams.operators.manifests;
 
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.EnvVar;
+import io.fabric8.kubernetes.api.model.EnvVarBuilder;
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.Namespace;
 import io.fabric8.kubernetes.api.model.NamespaceBuilder;
 import io.fabric8.kubernetes.api.model.Namespaced;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.api.model.rbac.ClusterRoleBinding;
 import io.fabric8.kubernetes.api.model.rbac.RoleBinding;
 import io.skodjob.testframe.TestFrameConstants;
@@ -20,6 +24,8 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -72,6 +78,8 @@ public static CompletableFuture<Void> install() throws IOException {
                 crb.getMetadata().setName(crb.getMetadata().getName() + "." + OPERATOR_NS);
             } else if (res instanceof RoleBinding rb) {
                 rb.getSubjects().forEach(sbj -> sbj.setNamespace(OPERATOR_NS));
+            } else if (res instanceof Deployment dep && DEPLOYMENT_NAME.equals(res.getMetadata().getName())) {
+                modifyDeployment(dep);
             }
             KubeResourceManager.getInstance().createOrUpdateResourceWithoutWait(res);
         });
@@ -80,6 +88,14 @@ public static CompletableFuture<Void> install() throws IOException {
             TestFrameConstants.GLOBAL_TIMEOUT, ApicurioRegistryManifestInstaller::isReady);
     }
 
+    private static void modifyDeployment(Deployment deployment) {
+        Container container = deployment.getSpec().getTemplate().getSpec().getContainers().get(0);
+        List<EnvVar> env = new ArrayList<>(container.getEnv() == null ? Collections.emptyList() : container.getEnv());
+        env.removeIf(envVar -> envVar.getName().equals("WATCH_NAMESPACE"));
+        env.add(new EnvVarBuilder().withName("WATCH_NAMESPACE").withValue("").build());
+        container.setEnv(env);
+    }
+
     private static boolean isReady() {
         if (KubeResourceManager.getKubeClient().getClient().apps()
             .deployments().inNamespace(OPERATOR_NS).withName(DEPLOYMENT_NAME).isReady()) {

src/main/java/io/streams/operators/manifests/FlinkManifestInstaller.java

@@ -8,7 +8,6 @@
 import io.fabric8.kubernetes.api.model.Namespace;
 import io.fabric8.kubernetes.api.model.NamespaceBuilder;
 import io.fabric8.kubernetes.api.model.Namespaced;
-import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.api.model.rbac.ClusterRoleBinding;
 import io.fabric8.kubernetes.api.model.rbac.RoleBinding;
 import io.skodjob.testframe.TestFrameConstants;
@@ -73,8 +72,6 @@ public static CompletableFuture<Void> install() throws IOException {
                 crb.getMetadata().setName(crb.getMetadata().getName() + "." + OPERATOR_NS);
             } else if (res instanceof RoleBinding rb) {
                 rb.getSubjects().forEach(sbj -> sbj.setNamespace(OPERATOR_NS));
-            } else if (res instanceof Deployment && DEPLOYMENT_NAME.equals(res.getMetadata().getName())) {
-                modifyDeployment((Deployment) res);
             } else {
                 res.getMetadata().setNamespace(OPERATOR_NS);
             }
@@ -85,10 +82,6 @@ public static CompletableFuture<Void> install() throws IOException {
             TestFrameConstants.GLOBAL_TIMEOUT, FlinkManifestInstaller::isReady);
     }
 
-    private static void modifyDeployment(Deployment deployment) {
-        deployment.getSpec().getTemplate().getSpec().setSecurityContext(null);
-    }
-
     private static boolean isReady() {
         if (KubeResourceManager.getKubeClient().getClient().apps()
             .deployments().inNamespace(OPERATOR_NS).withName(DEPLOYMENT_NAME).isReady()) {

src/test/java/io/streams/e2e/flink/sql/SqlExampleST.java

@@ -0,0 +1,34 @@
+package io.streams.e2e.flink.sql;
+
+import io.streams.e2e.Abstract;
+import io.streams.operators.manifests.ApicurioRegistryManifestInstaller;
+import io.streams.operators.manifests.CertManagerManifestInstaller;
+import io.streams.operators.manifests.FlinkManifestInstaller;
+import io.streams.operators.manifests.StrimziManifestInstaller;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+
+import static io.streams.constants.TestTags.SQL_EXAMPLE;
+
+@Tag(SQL_EXAMPLE)
+public class SqlExampleST extends Abstract {
+
+    @BeforeAll
+    void prepareOperators() throws IOException {
+        CompletableFuture.allOf(
+            StrimziManifestInstaller.install(),
+            CertManagerManifestInstaller.install(),
+            FlinkManifestInstaller.install(),
+            ApicurioRegistryManifestInstaller.install()
+        ).join();
+    }
+
+    @Test
+    void testFlinkSqlExample() {
+        //TODO
+    }
+}