improve

Signed-off-by: Attila Mészáros <[email protected]>

Author: csviri

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java

@@ -190,7 +190,7 @@ public void deploy(
         status.getJobStatus().setState(org.apache.flink.api.common.JobStatus.RECONCILING);
         status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
 
-        IngressUtils.updateIngressRules(ctx, spec, deployConfig, ctx.getKubernetesClient(), true);
+        IngressUtils.updateIngressRules(ctx, spec, deployConfig, ctx.getKubernetesClient());
     }
 
     private void setJobIdIfNecessary(

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java

@@ -106,7 +106,7 @@ public void deploy(
         setOwnerReference(cr, deployConfig);
         ctx.getFlinkService().submitSessionCluster(deployConfig);
         cr.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
-        IngressUtils.updateIngressRules(ctx, spec, deployConfig, ctx.getKubernetesClient(), false);
+        IngressUtils.updateIngressRules(ctx, spec, deployConfig, ctx.getKubernetesClient());
     }
 
     @Override

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java

@@ -17,7 +17,6 @@
 
 package org.apache.flink.kubernetes.operator.utils;
 
-import io.fabric8.kubernetes.api.model.networking.v1.Ingress;
 import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
 import org.apache.flink.kubernetes.operator.api.CrdConstants;
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
@@ -31,6 +30,7 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.api.model.networking.v1.Ingress;
 import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/IngressUtils.java

@@ -29,7 +29,6 @@
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.api.model.OwnerReference;
 import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder;
-import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.api.model.networking.v1.HTTPIngressRuleValueBuilder;
 import io.fabric8.kubernetes.api.model.networking.v1.IngressBuilder;
 import io.fabric8.kubernetes.api.model.networking.v1.IngressRule;
@@ -69,33 +68,17 @@ public static void updateIngressRules(
             FlinkResourceContext<?> ctx,
             FlinkDeploymentSpec spec,
             Configuration effectiveConfig,
-            KubernetesClient client,
-            boolean usePrimaryAsOwner) {
+            KubernetesClient client) {
 
         var objectMeta = ctx.getResource().getMetadata();
         if (spec.getIngress() != null) {
             HasMetadata ingress = getIngress(objectMeta, spec, effectiveConfig, client);
-            if (usePrimaryAsOwner) {
-                setOwnerReference(ctx.getResource(), Collections.singletonList(ingress));
-            } else {
-                Deployment deployment =
-                        client.apps()
-                                .deployments()
-                                .inNamespace(objectMeta.getNamespace())
-                                .withName(objectMeta.getName())
-                                .get();
-                if (deployment == null) {
-                    LOG.error("Could not find deployment {}", objectMeta.getName());
-                } else {
-                    setOwnerReference(deployment, Collections.singletonList(ingress));
-                }
-            }
-
+            setOwnerReference(ctx.getResource(), Collections.singletonList(ingress));
             LOG.info("Updating ingress rules {}", ingress);
             client.resource(ingress)
                     .inNamespace(objectMeta.getNamespace())
                     .createOr(NonDeletingOperation::update);
-        } else if (usePrimaryAsOwner) {
+        } else {
             Optional<? extends HasMetadata> ingress;
             if (ingressInNetworkingV1(client)) {
                 ingress =