You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/05/13 13:49:43 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27337] Prevent session cluster to be deleted when there are running jobs
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 5239cd6 [FLINK-27337] Prevent session cluster to be deleted when there are running jobs
5239cd6 is described below
commit 5239cd6a7ae0812e1c399949f0facde972dcfa78
Author: Aitozi <yu...@alibaba-inc.com>
AuthorDate: Wed May 11 12:31:29 2022 +0800
[FLINK-27337] Prevent session cluster to be deleted when there are running jobs
---
.../flink/kubernetes/operator/FlinkOperator.java | 12 ++--
.../operator/controller/FlinkControllerConfig.java | 5 ++
.../controller/FlinkDeploymentController.java | 20 +++---
.../controller/FlinkSessionJobController.java | 71 ++++------------------
.../operator/informer/InformerManager.java | 67 ++++++++++++++++++++
.../reconciler/deployment/ReconcilerFactory.java | 8 ++-
.../reconciler/deployment/SessionReconciler.java | 37 ++++++++++-
.../kubernetes/operator/utils/OperatorUtils.java | 43 +++++++++++++
.../flink/kubernetes/operator/TestUtils.java | 26 ++++----
.../controller/FlinkDeploymentControllerTest.java | 5 +-
.../deployment/SessionReconcilerTest.java | 6 +-
11 files changed, 209 insertions(+), 91 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index ce4a61e..6cb961f 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
@@ -31,6 +31,7 @@ import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus;
+import org.apache.flink.kubernetes.operator.informer.InformerManager;
import org.apache.flink.kubernetes.operator.metrics.MetricManager;
import org.apache.flink.kubernetes.operator.metrics.OperatorMetricUtils;
import org.apache.flink.kubernetes.operator.observer.Observer;
@@ -72,6 +73,7 @@ public class FlinkOperator {
private final FlinkConfigManager configManager;
private final Set<FlinkResourceValidator> validators;
private final MetricGroup metricGroup;
+ private final InformerManager informerManager;
public FlinkOperator(@Nullable Configuration conf) {
this.client = new DefaultKubernetesClient();
@@ -83,6 +85,9 @@ public class FlinkOperator {
this.validators = ValidatorUtils.discoverValidators(configManager);
this.metricGroup =
OperatorMetricUtils.initOperatorMetrics(configManager.getDefaultConfig());
+ Set<String> watchedNamespaces =
+ configManager.getOperatorConfiguration().getWatchedNamespaces();
+ this.informerManager = new InformerManager(watchedNamespaces, client);
PluginManager pluginManager =
PluginUtils.createPluginManagerFromRootFolder(configManager.getDefaultConfig());
FileSystem.initialize(configManager.getDefaultConfig(), pluginManager);
@@ -91,7 +96,7 @@ public class FlinkOperator {
private void registerDeploymentController() {
StatusHelper<FlinkDeploymentStatus> statusHelper = new StatusHelper<>(client);
ReconcilerFactory reconcilerFactory =
- new ReconcilerFactory(client, flinkService, configManager);
+ new ReconcilerFactory(client, flinkService, configManager, informerManager);
ObserverFactory observerFactory =
new ObserverFactory(client, flinkService, configManager, statusHelper);
@@ -109,7 +114,6 @@ public class FlinkOperator {
new FlinkControllerConfig<>(
controller,
configManager.getOperatorConfiguration().getWatchedNamespaces());
- controller.setControllerConfig(controllerConfig);
controllerConfig.setConfigurationService(configurationService);
operator.register(controller, controllerConfig);
}
@@ -128,7 +132,8 @@ public class FlinkOperator {
reconciler,
observer,
new MetricManager<>(metricGroup),
- statusHelper);
+ statusHelper,
+ informerManager);
FlinkControllerConfig<FlinkSessionJob> controllerConfig =
new FlinkControllerConfig<>(
@@ -136,7 +141,6 @@ public class FlinkOperator {
configManager.getOperatorConfiguration().getWatchedNamespaces());
controllerConfig.setConfigurationService(configurationService);
- controller.init(controllerConfig);
operator.register(controller, controllerConfig);
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkControllerConfig.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkControllerConfig.java
index 2a55456..fb370ee 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkControllerConfig.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkControllerConfig.java
@@ -37,4 +37,9 @@ public class FlinkControllerConfig<CR extends HasMetadata> extends AnnotationCon
public Set<String> getNamespaces() {
return watchedNamespaces;
}
+
+ @Override
+ public Set<String> getEffectiveNamespaces() {
+ return watchedNamespaces;
+ }
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
index d38e665..32651cd 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
@@ -17,7 +17,9 @@
package org.apache.flink.kubernetes.operator.controller;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
@@ -31,7 +33,6 @@ import org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFact
import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
import org.apache.flink.kubernetes.operator.utils.StatusHelper;
import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
-import org.apache.flink.util.Preconditions;
import io.fabric8.kubernetes.api.model.Event;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -51,6 +52,7 @@ import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
/** Controller that runs the main reconcile loop for Flink deployments. */
@@ -68,8 +70,10 @@ public class FlinkDeploymentController
private final ReconcilerFactory reconcilerFactory;
private final ObserverFactory observerFactory;
private final MetricManager<FlinkDeployment> metricManager;
- private FlinkControllerConfig<FlinkDeployment> controllerConfig;
private final StatusHelper<FlinkDeploymentStatus> statusHelper;
+ private Set<String> effectiveNamespaces;
+ private final ConcurrentHashMap<Tuple2<String, String>, FlinkDeploymentStatus> statusCache =
+ new ConcurrentHashMap<>();
public FlinkDeploymentController(
FlinkConfigManager configManager,
@@ -86,6 +90,7 @@ public class FlinkDeploymentController
this.observerFactory = observerFactory;
this.metricManager = metricManager;
this.statusHelper = statusHelper;
+ this.effectiveNamespaces = configManager.getOperatorConfiguration().getWatchedNamespaces();
}
@Override
@@ -145,8 +150,6 @@ public class FlinkDeploymentController
@Override
public List<EventSource> prepareEventSources(EventSourceContext<FlinkDeployment> ctx) {
- Preconditions.checkNotNull(controllerConfig, "Controller config cannot be null");
- Set<String> effectiveNamespaces = controllerConfig.getEffectiveNamespaces();
if (effectiveNamespaces.isEmpty()) {
return List.of(OperatorUtils.createJmDepInformerEventSource(kubernetesClient));
} else {
@@ -156,6 +159,11 @@ public class FlinkDeploymentController
}
}
+ @VisibleForTesting
+ public void setEffectiveNamespaces(Set<String> effectiveNamespaces) {
+ this.effectiveNamespaces = effectiveNamespaces;
+ }
+
@Override
public Optional<FlinkDeployment> updateErrorStatus(
FlinkDeployment flinkApp, RetryInfo retryInfo, RuntimeException e) {
@@ -163,10 +171,6 @@ public class FlinkDeploymentController
flinkApp, retryInfo, e, metricManager, statusHelper);
}
- public void setControllerConfig(FlinkControllerConfig<FlinkDeployment> config) {
- this.controllerConfig = config;
- }
-
private boolean validateDeployment(FlinkDeployment deployment) {
for (FlinkResourceValidator validator : validators) {
Optional<String> validationError = validator.validateDeployment(deployment);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
index 86be377..586fc6a 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobController.java
@@ -22,6 +22,7 @@ import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus;
import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.informer.InformerManager;
import org.apache.flink.kubernetes.operator.metrics.MetricManager;
import org.apache.flink.kubernetes.operator.observer.Observer;
import org.apache.flink.kubernetes.operator.reconciler.Reconciler;
@@ -29,12 +30,10 @@ import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
import org.apache.flink.kubernetes.operator.utils.StatusHelper;
import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
-import org.apache.flink.util.Preconditions;
import io.fabric8.kubernetes.api.model.KubernetesResourceList;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
-import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
@@ -50,13 +49,10 @@ import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEven
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.function.Function;
import java.util.stream.Collectors;
/** Controller that runs the main reconcile loop for {@link FlinkSessionJob}. */
@@ -67,8 +63,6 @@ public class FlinkSessionJobController
EventSourceInitializer<FlinkSessionJob> {
private static final Logger LOG = LoggerFactory.getLogger(FlinkSessionJobController.class);
- private static final String CLUSTER_ID_INDEX = "clusterId_index";
- private static final String ALL_NAMESPACE = "allNamespace";
private final FlinkConfigManager configManager;
private final KubernetesClient kubernetesClient;
@@ -77,8 +71,8 @@ public class FlinkSessionJobController
private final Reconciler<FlinkSessionJob> reconciler;
private final Observer<FlinkSessionJob> observer;
private final MetricManager<FlinkSessionJob> metricManager;
- private Map<String, SharedIndexInformer<FlinkSessionJob>> informers;
- private FlinkControllerConfig<FlinkSessionJob> controllerConfig;
+ private final InformerManager informerManager;
+ private final Set<String> effectiveNamespaces;
private final StatusHelper<FlinkSessionJobStatus> statusHelper;
@@ -89,7 +83,8 @@ public class FlinkSessionJobController
Reconciler<FlinkSessionJob> reconciler,
Observer<FlinkSessionJob> observer,
MetricManager<FlinkSessionJob> metricManager,
- StatusHelper<FlinkSessionJobStatus> statusHelper) {
+ StatusHelper<FlinkSessionJobStatus> statusHelper,
+ InformerManager informerManager) {
this.configManager = configManager;
this.kubernetesClient = kubernetesClient;
this.validators = validators;
@@ -97,11 +92,8 @@ public class FlinkSessionJobController
this.observer = observer;
this.metricManager = metricManager;
this.statusHelper = statusHelper;
- }
-
- public void init(FlinkControllerConfig<FlinkSessionJob> config) {
- this.controllerConfig = config;
- this.informers = createInformers();
+ this.informerManager = informerManager;
+ this.effectiveNamespaces = configManager.getOperatorConfiguration().getWatchedNamespaces();
}
@Override
@@ -118,7 +110,6 @@ public class FlinkSessionJobController
}
try {
- // TODO refactor the reconciler interface to return UpdateControl directly
reconciler.reconcile(flinkSessionJob, context);
} catch (Exception e) {
throw new ReconciliationException(e);
@@ -147,13 +138,11 @@ public class FlinkSessionJobController
@Override
public List<EventSource> prepareEventSources(
EventSourceContext<FlinkSessionJob> eventSourceContext) {
- Preconditions.checkNotNull(controllerConfig, "Controller config cannot be null");
- Set<String> effectiveNamespaces = controllerConfig.getEffectiveNamespaces();
if (effectiveNamespaces.isEmpty()) {
return List.of(
createFlinkDepInformerEventSource(
kubernetesClient.resources(FlinkDeployment.class).inAnyNamespace(),
- ALL_NAMESPACE));
+ OperatorUtils.ALL_NAMESPACE));
} else {
return effectiveNamespaces.stream()
.map(
@@ -195,14 +184,13 @@ public class FlinkSessionJobController
private PrimaryResourcesRetriever<FlinkDeployment> primaryResourceRetriever() {
return flinkDeployment -> {
var namespace = flinkDeployment.getMetadata().getNamespace();
- var informer =
- controllerConfig.getEffectiveNamespaces().isEmpty()
- ? informers.get(ALL_NAMESPACE)
- : informers.get(namespace);
+ var informer = informerManager.getSessionJobInformer(namespace);
var sessionJobs =
informer.getIndexer()
- .byIndex(CLUSTER_ID_INDEX, flinkDeployment.getMetadata().getName());
+ .byIndex(
+ OperatorUtils.CLUSTER_ID_INDEX,
+ flinkDeployment.getMetadata().getName());
var resourceIDs = new HashSet<ResourceID>();
for (FlinkSessionJob sessionJob : sessionJobs) {
resourceIDs.add(
@@ -218,41 +206,6 @@ public class FlinkSessionJobController
};
}
- /**
- * Create informers for session job to build indexer for cluster to session job relations.
- *
- * @return The different namespace's index informer.
- */
- private Map<String, SharedIndexInformer<FlinkSessionJob>> createInformers() {
- Set<String> effectiveNamespaces = controllerConfig.getEffectiveNamespaces();
- if (effectiveNamespaces.isEmpty()) {
- return Map.of(
- ALL_NAMESPACE,
- kubernetesClient
- .resources(FlinkSessionJob.class)
- .inAnyNamespace()
- .withIndexers(clusterToSessionJobIndexer())
- .inform());
- } else {
- var informers = new HashMap<String, SharedIndexInformer<FlinkSessionJob>>();
- for (String effectiveNamespace : effectiveNamespaces) {
- informers.put(
- effectiveNamespace,
- kubernetesClient
- .resources(FlinkSessionJob.class)
- .inNamespace(effectiveNamespace)
- .withIndexers(clusterToSessionJobIndexer())
- .inform());
- }
- return informers;
- }
- }
-
- private Map<String, Function<FlinkSessionJob, List<String>>> clusterToSessionJobIndexer() {
- return Map.of(
- CLUSTER_ID_INDEX, sessionJob -> List.of(sessionJob.getSpec().getDeploymentName()));
- }
-
private boolean validateSessionJob(FlinkSessionJob sessionJob, Context context) {
for (FlinkResourceValidator validator : validators) {
Optional<String> validationError =
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/informer/InformerManager.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/informer/InformerManager.java
new file mode 100644
index 0000000..0f8d86d
--- /dev/null
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/informer/InformerManager.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.informer;
+
+import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
+import org.apache.flink.util.Preconditions;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Set;
+
+/** The manager of the informers. */
+public class InformerManager {
+
+ private static final Logger LOG = LoggerFactory.getLogger(InformerManager.class);
+ private final Set<String> watchedNamespaces;
+ private final KubernetesClient kubernetesClient;
+ private volatile Map<String, SharedIndexInformer<FlinkSessionJob>> sessionJobInformers;
+
+ public InformerManager(Set<String> watchedNamespaces, KubernetesClient kubernetesClient) {
+ this.watchedNamespaces = watchedNamespaces;
+ this.kubernetesClient = kubernetesClient;
+ }
+
+ public SharedIndexInformer<FlinkSessionJob> getSessionJobInformer(String namespace) {
+ initSessionJobInformer();
+ var effectiveNamespace =
+ watchedNamespaces.isEmpty() ? OperatorUtils.ALL_NAMESPACE : namespace;
+ var informer = sessionJobInformers.get(effectiveNamespace);
+ Preconditions.checkNotNull(
+ informer, String.format("The informer for %s should not be null", namespace));
+ return informer;
+ }
+
+ private void initSessionJobInformer() {
+ if (sessionJobInformers == null) {
+ synchronized (this) {
+ if (sessionJobInformers == null) {
+ this.sessionJobInformers =
+ OperatorUtils.createSessionJobInformersWithIndexer(
+ watchedNamespaces, kubernetesClient);
+ LOG.info("Created informer for {}", sessionJobInformers.keySet());
+ }
+ }
+ }
+ }
+}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ReconcilerFactory.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ReconcilerFactory.java
index 7603469..c5140eb 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ReconcilerFactory.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ReconcilerFactory.java
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.operator.reconciler.deployment;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.config.Mode;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.informer.InformerManager;
import org.apache.flink.kubernetes.operator.reconciler.Reconciler;
import org.apache.flink.kubernetes.operator.service.FlinkService;
@@ -35,15 +36,18 @@ public class ReconcilerFactory {
private final FlinkService flinkService;
private final FlinkConfigManager configManager;
private final Map<Mode, Reconciler<FlinkDeployment>> reconcilerMap;
+ private final InformerManager informerManager;
public ReconcilerFactory(
KubernetesClient kubernetesClient,
FlinkService flinkService,
- FlinkConfigManager configManager) {
+ FlinkConfigManager configManager,
+ InformerManager informerManager) {
this.kubernetesClient = kubernetesClient;
this.flinkService = flinkService;
this.configManager = configManager;
this.reconcilerMap = new ConcurrentHashMap<>();
+ this.informerManager = informerManager;
}
public Reconciler<FlinkDeployment> getOrCreate(FlinkDeployment flinkApp) {
@@ -53,7 +57,7 @@ public class ReconcilerFactory {
switch (mode) {
case SESSION:
return new SessionReconciler(
- kubernetesClient, flinkService, configManager);
+ kubernetesClient, flinkService, configManager, informerManager);
case APPLICATION:
return new ApplicationReconciler(
kubernetesClient, flinkService, configManager);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
index c4e23e6..4490784 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
@@ -25,16 +25,21 @@ import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
import org.apache.flink.kubernetes.operator.crd.status.ReconciliationStatus;
+import org.apache.flink.kubernetes.operator.informer.InformerManager;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.operator.utils.IngressUtils;
+import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.stream.Collectors;
+
/**
* Reconciler responsible for handling the session cluster lifecycle according to the desired and
* current states.
@@ -42,12 +47,15 @@ import org.slf4j.LoggerFactory;
public class SessionReconciler extends AbstractDeploymentReconciler {
private static final Logger LOG = LoggerFactory.getLogger(SessionReconciler.class);
+ private final InformerManager informerManager;
public SessionReconciler(
KubernetesClient kubernetesClient,
FlinkService flinkService,
- FlinkConfigManager configManager) {
+ FlinkConfigManager configManager,
+ InformerManager informerManager) {
super(kubernetesClient, flinkService, configManager);
+ this.informerManager = informerManager;
}
@Override
@@ -134,4 +142,31 @@ public class SessionReconciler extends AbstractDeploymentReconciler {
flinkService.deleteClusterDeployment(
deployment.getMetadata(), deployment.getStatus(), true);
}
+
+ @Override
+ public DeleteControl cleanup(FlinkDeployment flinkApp, Context context) {
+ var sessionJobs =
+ informerManager
+ .getSessionJobInformer(flinkApp.getMetadata().getNamespace())
+ .getIndexer()
+ .byIndex(OperatorUtils.CLUSTER_ID_INDEX, flinkApp.getMetadata().getName());
+ if (!sessionJobs.isEmpty()) {
+ var error =
+ String.format(
+ "The session jobs %s should be deleted first",
+ sessionJobs.stream()
+ .map(job -> job.getMetadata().getName())
+ .collect(Collectors.toList()));
+ // TODO generate error events for this
+ LOG.warn(error);
+ return DeleteControl.noFinalizerRemoval()
+ .rescheduleAfter(
+ configManager
+ .getOperatorConfiguration()
+ .getReconcileInterval()
+ .toMillis());
+ } else {
+ return super.cleanup(flinkApp, context);
+ }
+ }
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/OperatorUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/OperatorUtils.java
index d76bb03..5844522 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/OperatorUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/OperatorUtils.java
@@ -38,9 +38,13 @@ import org.slf4j.LoggerFactory;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Function;
/** Operator SDK related utility functions. */
public class OperatorUtils {
@@ -48,6 +52,8 @@ public class OperatorUtils {
private static final Logger LOG = LoggerFactory.getLogger(OperatorUtils.class);
private static final String NAMESPACES_SPLITTER_KEY = ",";
+ public static final String CLUSTER_ID_INDEX = "clusterId_index";
+ public static final String ALL_NAMESPACE = "allNamespace";
public static InformerEventSource<Deployment, HasMetadata> createJmDepInformerEventSource(
KubernetesClient kubernetesClient, String namespace) {
@@ -99,4 +105,41 @@ public class OperatorUtils {
: null;
return context.getSecondaryResource(FlinkDeployment.class, identifier);
}
+
+ /**
+ * Create informers for session job to build indexer for cluster to session job relations.
+ *
+ * @return The different namespace's index informer.
+ */
+ public static Map<String, SharedIndexInformer<FlinkSessionJob>>
+ createSessionJobInformersWithIndexer(
+ Set<String> effectiveNamespaces, KubernetesClient kubernetesClient) {
+ if (effectiveNamespaces.isEmpty()) {
+ return Map.of(
+ ALL_NAMESPACE,
+ kubernetesClient
+ .resources(FlinkSessionJob.class)
+ .inAnyNamespace()
+ .withIndexers(clusterToSessionJobIndexer())
+ .inform());
+ } else {
+ var informers = new HashMap<String, SharedIndexInformer<FlinkSessionJob>>();
+ for (String effectiveNamespace : effectiveNamespaces) {
+ informers.put(
+ effectiveNamespace,
+ kubernetesClient
+ .resources(FlinkSessionJob.class)
+ .inNamespace(effectiveNamespace)
+ .withIndexers(clusterToSessionJobIndexer())
+ .inform());
+ }
+ return informers;
+ }
+ }
+
+ private static Map<String, Function<FlinkSessionJob, List<String>>>
+ clusterToSessionJobIndexer() {
+ return Map.of(
+ CLUSTER_ID_INDEX, sessionJob -> List.of(sessionJob.getSpec().getDeploymentName()));
+ }
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
index 21ee51e..38ea772 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
@@ -22,7 +22,6 @@ import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
-import org.apache.flink.kubernetes.operator.controller.FlinkControllerConfig;
import org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
@@ -39,6 +38,7 @@ import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
+import org.apache.flink.kubernetes.operator.informer.InformerManager;
import org.apache.flink.kubernetes.operator.metrics.MetricManager;
import org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
import org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFactory;
@@ -71,6 +71,7 @@ import java.lang.reflect.Field;
import java.net.HttpURLConnection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -366,19 +367,18 @@ public class TestUtils {
TestingFlinkService flinkService) {
var statusHelper = new StatusHelper<FlinkDeploymentStatus>(kubernetesClient);
- var controller =
- new FlinkDeploymentController(
- configManager,
+ return new FlinkDeploymentController(
+ configManager,
+ kubernetesClient,
+ ValidatorUtils.discoverValidators(configManager),
+ new ReconcilerFactory(
kubernetesClient,
- ValidatorUtils.discoverValidators(configManager),
- new ReconcilerFactory(kubernetesClient, flinkService, configManager),
- new ObserverFactory(
- kubernetesClient, flinkService, configManager, statusHelper),
- new MetricManager<>(new MetricListener().getMetricGroup()),
- statusHelper);
- controller.setControllerConfig(
- new FlinkControllerConfig(controller, Collections.emptySet()));
- return controller;
+ flinkService,
+ configManager,
+ new InformerManager(new HashSet<>(), kubernetesClient)),
+ new ObserverFactory(kubernetesClient, flinkService, configManager, statusHelper),
+ new MetricManager<>(new MetricListener().getMetricGroup()),
+ statusHelper);
}
/** Testing ResponseProvider. */
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
index 5095afd..e40c680 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
@@ -629,15 +629,14 @@ public class FlinkDeploymentControllerTest {
@Test
public void testPrepareEventSource() {
// Test watch all
- testController.setControllerConfig(
- new FlinkControllerConfig(testController, Collections.emptySet()));
+ testController.setEffectiveNamespaces(Collections.emptySet());
List<EventSource> eventSources = testController.prepareEventSources(null);
assertEquals(1, eventSources.size());
assertEquals("all", eventSources.get(0).name());
// Test watch namespaces
Set<String> namespaces = Set.of("ns1", "ns2", "ns3");
- testController.setControllerConfig(new FlinkControllerConfig(testController, namespaces));
+ testController.setEffectiveNamespaces(namespaces);
eventSources = testController.prepareEventSources(null);
assertEquals(3, eventSources.size());
assertEquals(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
index 0f7cfbf..1c5ea67 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconcilerTest.java
@@ -22,10 +22,12 @@ import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.informer.InformerManager;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import org.junit.jupiter.api.Test;
+import java.util.HashSet;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -36,6 +38,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
public class SessionReconcilerTest {
private final FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
+ private final InformerManager informerManager = new InformerManager(new HashSet<>(), null);
@Test
public void testStartSession() throws Exception {
@@ -50,7 +53,8 @@ public class SessionReconcilerTest {
}
};
- SessionReconciler reconciler = new SessionReconciler(null, flinkService, configManager);
+ SessionReconciler reconciler =
+ new SessionReconciler(null, flinkService, configManager, informerManager);
FlinkDeployment deployment = TestUtils.buildSessionCluster();
reconciler.reconcile(deployment, context);
assertEquals(1, count.get());