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 d2f1d71 [FLINK-27714] Migrate to java-operator-sdk v3
d2f1d71 is described below
commit d2f1d71c2c28d566c3451f2878a301130ce63807
Author: Matyas Orhidi <[email protected]>
AuthorDate: Mon May 23 22:33:17 2022 +0200
[FLINK-27714] Migrate to java-operator-sdk v3
---
.../flink/kubernetes/operator/FlinkOperator.java | 65 ++++-------
.../operator/config/FlinkConfigManager.java | 11 +-
.../operator/controller/FlinkControllerConfig.java | 5 +-
.../controller/FlinkDeploymentController.java | 45 +++----
.../controller/FlinkSessionJobController.java | 118 +++----------------
.../operator/informer/InformerManager.java | 81 ++++++-------
.../deployment/AbstractDeploymentObserver.java | 10 +-
.../observer/sessionjob/SessionJobObserver.java | 6 +-
.../operator/reconciler/ReconciliationUtils.java | 22 ++--
.../reconciler/deployment/ReconcilerFactory.java | 8 +-
.../reconciler/deployment/SessionReconciler.java | 16 +--
.../sessionjob/FlinkSessionJobReconciler.java | 7 +-
.../flink/kubernetes/operator/utils/EnvUtils.java | 16 +++
.../operator/utils/EventSourceUtils.java | 116 ++++++++++++++++++
.../kubernetes/operator/utils/OperatorUtils.java | 129 ---------------------
.../kubernetes/operator/FlinkOperatorTest.java | 26 ++---
.../flink/kubernetes/operator/TestUtils.java | 128 ++++++++++++++++----
.../kubernetes/operator/TestingFlinkService.java | 21 +++-
.../controller/DeploymentRecoveryTest.java | 5 +-
.../controller/FlinkDeploymentControllerTest.java | 40 ++-----
.../operator/informer/InformerManagerTest.java | 13 +--
.../observer/deployment/SessionObserverTest.java | 83 -------------
.../deployment/SessionReconcilerTest.java | 6 +-
pom.xml | 3 +-
24 files changed, 407 insertions(+), 573 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 6cb961f..4770617 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,7 +31,6 @@ 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;
@@ -50,9 +49,7 @@ import org.apache.flink.metrics.MetricGroup;
import io.fabric8.kubernetes.client.DefaultKubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.Operator;
-import io.javaoperatorsdk.operator.api.config.ConfigurationService;
import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider;
-import io.javaoperatorsdk.operator.config.runtime.DefaultConfigurationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,6 +57,7 @@ import javax.annotation.Nullable;
import java.util.Set;
import java.util.concurrent.Executors;
+import java.util.function.Consumer;
/** Main Class for Flink native k8s operator. */
public class FlinkOperator {
@@ -69,34 +67,47 @@ public class FlinkOperator {
private final KubernetesClient client;
private final FlinkService flinkService;
- private final ConfigurationService configurationService;
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();
this.configManager = conf != null ? new FlinkConfigManager(conf) : new
FlinkConfigManager();
- this.configurationService =
-
getConfigurationService(configManager.getOperatorConfiguration());
- this.operator = new Operator(client, configurationService);
+ this.operator =
+ new Operator(
+ client,
+ getConfigurationServiceOverriderConsumer(
+ configManager.getOperatorConfiguration()));
this.flinkService = new FlinkService(client, configManager);
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);
}
+ @VisibleForTesting
+ protected static Consumer<ConfigurationServiceOverrider>
+ getConfigurationServiceOverriderConsumer(
+ FlinkOperatorConfiguration operatorConfiguration) {
+ return overrider -> {
+ int parallelism =
operatorConfiguration.getReconcilerMaxParallelism();
+ if (parallelism == -1) {
+ LOG.info("Configuring operator with unbounded reconciliation
thread pool.");
+ overrider.withExecutorService(Executors.newCachedThreadPool());
+ } else {
+ LOG.info("Configuring operator with {} reconciliation
threads.", parallelism);
+ overrider.withConcurrentReconciliationThreads(parallelism);
+ }
+ };
+ }
+
private void registerDeploymentController() {
StatusHelper<FlinkDeploymentStatus> statusHelper = new
StatusHelper<>(client);
ReconcilerFactory reconcilerFactory =
- new ReconcilerFactory(client, flinkService, configManager,
informerManager);
+ new ReconcilerFactory(client, flinkService, configManager);
ObserverFactory observerFactory =
new ObserverFactory(client, flinkService, configManager,
statusHelper);
@@ -114,7 +125,6 @@ public class FlinkOperator {
new FlinkControllerConfig<>(
controller,
configManager.getOperatorConfiguration().getWatchedNamespaces());
- controllerConfig.setConfigurationService(configurationService);
operator.register(controller, controllerConfig);
}
@@ -127,41 +137,19 @@ public class FlinkOperator {
FlinkSessionJobController controller =
new FlinkSessionJobController(
configManager,
- client,
validators,
reconciler,
observer,
new MetricManager<>(metricGroup),
- statusHelper,
- informerManager);
+ statusHelper);
FlinkControllerConfig<FlinkSessionJob> controllerConfig =
new FlinkControllerConfig<>(
controller,
configManager.getOperatorConfiguration().getWatchedNamespaces());
-
- controllerConfig.setConfigurationService(configurationService);
operator.register(controller, controllerConfig);
}
- private ConfigurationService getConfigurationService(
- FlinkOperatorConfiguration operatorConfiguration) {
-
- ConfigurationServiceOverrider configOverrider =
- new
ConfigurationServiceOverrider(DefaultConfigurationService.instance())
- .checkingCRDAndValidateLocalModel(false);
-
- int parallelism = operatorConfiguration.getReconcilerMaxParallelism();
- if (parallelism == -1) {
- LOG.info("Configuring operator with unbounded reconciliation
thread pool.");
- configOverrider =
configOverrider.withExecutorService(Executors.newCachedThreadPool());
- } else {
- LOG.info("Configuring operator with {} reconciliation threads.",
parallelism);
- configOverrider =
configOverrider.withConcurrentReconciliationThreads(parallelism);
- }
- return configOverrider.build();
- }
-
public void run() {
registerDeploymentController();
registerSessionJobController();
@@ -169,11 +157,6 @@ public class FlinkOperator {
operator.start();
}
- @VisibleForTesting
- protected Operator getOperator() {
- return operator;
- }
-
public static void main(String... args) {
EnvUtils.logEnvironmentInfo(LOG, "Flink Kubernetes Operator", args);
new FlinkOperator(null).run();
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
index f4c7a0b..3a7c1c3 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
@@ -25,7 +25,7 @@ import
org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
-import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
+import org.apache.flink.kubernetes.operator.utils.EnvUtils;
import org.apache.flink.shaded.guava30.com.google.common.cache.Cache;
import org.apache.flink.shaded.guava30.com.google.common.cache.CacheBuilder;
@@ -63,7 +63,7 @@ public class FlinkConfigManager {
private final AtomicLong defaultConfigVersion = new AtomicLong(0);
private final LoadingCache<Key, Configuration> cache;
- private Set<String> namespaces = OperatorUtils.getWatchedNamespaces();
+ private final Set<String> namespaces = EnvUtils.getWatchedNamespaces();
public FlinkConfigManager() {
this(GlobalConfiguration.loadConfiguration());
@@ -188,13 +188,6 @@ public class FlinkConfigManager {
LOG.info("Enabled dynamic config updates, checking config changes
every {}", checkInterval);
}
- @VisibleForTesting
- public void setWatchedNamespaces(Set<String> namespaces) {
- this.namespaces = namespaces;
- operatorConfiguration =
- FlinkOperatorConfiguration.fromConfiguration(defaultConfig,
namespaces);
- }
-
@VisibleForTesting
protected Cache<Key, Configuration> getCache() {
return cache;
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 fb370ee..044780b 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
@@ -19,12 +19,13 @@ package org.apache.flink.kubernetes.operator.controller;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
-import io.javaoperatorsdk.operator.config.runtime.AnnotationConfiguration;
+import
io.javaoperatorsdk.operator.config.runtime.AnnotationControllerConfiguration;
import java.util.Set;
/** Custom config for {@link FlinkDeploymentController}. */
-public class FlinkControllerConfig<CR extends HasMetadata> extends
AnnotationConfiguration<CR> {
+public class FlinkControllerConfig<CR extends HasMetadata>
+ extends AnnotationControllerConfiguration<CR> {
private final Set<String> 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 432c8c1..439024c 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,6 @@
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;
@@ -30,37 +29,38 @@ import
org.apache.flink.kubernetes.operator.metrics.MetricManager;
import
org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import
org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFactory;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
import org.apache.flink.kubernetes.operator.utils.EventUtils;
-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 io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
-import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
import io.javaoperatorsdk.operator.processing.event.source.EventSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.List;
+import java.util.Map;
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. */
-@ControllerConfiguration
+@ControllerConfiguration()
public class FlinkDeploymentController
implements Reconciler<FlinkDeployment>,
ErrorStatusHandler<FlinkDeployment>,
- EventSourceInitializer<FlinkDeployment> {
+ EventSourceInitializer<FlinkDeployment>,
+ Cleaner<FlinkDeployment> {
private static final Logger LOG =
LoggerFactory.getLogger(FlinkDeploymentController.class);
private final FlinkConfigManager configManager;
@@ -71,7 +71,6 @@ public class FlinkDeploymentController
private final ObserverFactory observerFactory;
private final MetricManager<FlinkDeployment> metricManager;
private final StatusHelper<FlinkDeploymentStatus> statusHelper;
- private Set<String> effectiveNamespaces;
private final ConcurrentHashMap<Tuple2<String, String>,
FlinkDeploymentStatus> statusCache =
new ConcurrentHashMap<>();
@@ -90,7 +89,6 @@ public class FlinkDeploymentController
this.observerFactory = observerFactory;
this.metricManager = metricManager;
this.statusHelper = statusHelper;
- this.effectiveNamespaces =
configManager.getOperatorConfiguration().getWatchedNamespaces();
}
@Override
@@ -108,7 +106,8 @@ public class FlinkDeploymentController
}
@Override
- public UpdateControl<FlinkDeployment> reconcile(FlinkDeployment flinkApp,
Context context) {
+ public UpdateControl<FlinkDeployment> reconcile(FlinkDeployment flinkApp,
Context context)
+ throws Exception {
LOG.info("Starting reconciliation");
statusHelper.updateStatusFromCache(flinkApp);
FlinkDeployment previousDeployment =
ReconciliationUtils.clone(flinkApp);
@@ -152,26 +151,18 @@ public class FlinkDeploymentController
}
@Override
- public List<EventSource>
prepareEventSources(EventSourceContext<FlinkDeployment> ctx) {
- if (effectiveNamespaces.isEmpty()) {
- return
List.of(OperatorUtils.createJmDepInformerEventSource(kubernetesClient));
- } else {
- return effectiveNamespaces.stream()
- .map(ns ->
OperatorUtils.createJmDepInformerEventSource(kubernetesClient, ns))
- .collect(Collectors.toList());
- }
- }
-
- @VisibleForTesting
- public void setEffectiveNamespaces(Set<String> effectiveNamespaces) {
- this.effectiveNamespaces = effectiveNamespaces;
+ public Map<String, EventSource> prepareEventSources(
+ EventSourceContext<FlinkDeployment> context) {
+ return EventSourceInitializer.nameEventSources(
+ EventSourceUtils.getSessionJobInformerEventSource(context),
+ EventSourceUtils.getDeploymentInformerEventSource(context));
}
@Override
- public Optional<FlinkDeployment> updateErrorStatus(
- FlinkDeployment flinkApp, RetryInfo retryInfo, RuntimeException e)
{
- return ReconciliationUtils.updateErrorStatus(
- flinkApp, retryInfo, e, metricManager, statusHelper);
+ public ErrorStatusUpdateControl<FlinkDeployment> updateErrorStatus(
+ FlinkDeployment flinkDeployment, Context<FlinkDeployment> context,
Exception e) {
+ return ReconciliationUtils.toErrorStatusUpdateControl(
+ flinkDeployment, context.getRetryInfo(), e, metricManager,
statusHelper);
}
private boolean validateDeployment(FlinkDeployment 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 f3d3a52..a4b21f4 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,84 +22,68 @@ 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;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
-import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
import org.apache.flink.kubernetes.operator.utils.StatusHelper;
import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
-import io.fabric8.kubernetes.api.model.KubernetesResourceList;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
-import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
-import io.javaoperatorsdk.operator.processing.event.ResourceID;
import io.javaoperatorsdk.operator.processing.event.source.EventSource;
-import
io.javaoperatorsdk.operator.processing.event.source.PrimaryResourcesRetriever;
-import
io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.HashSet;
-import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.stream.Collectors;
/** Controller that runs the main reconcile loop for {@link FlinkSessionJob}.
*/
-@ControllerConfiguration
+@ControllerConfiguration()
public class FlinkSessionJobController
implements
io.javaoperatorsdk.operator.api.reconciler.Reconciler<FlinkSessionJob>,
ErrorStatusHandler<FlinkSessionJob>,
- EventSourceInitializer<FlinkSessionJob> {
+ EventSourceInitializer<FlinkSessionJob>,
+ Cleaner<FlinkSessionJob> {
private static final Logger LOG =
LoggerFactory.getLogger(FlinkSessionJobController.class);
private final FlinkConfigManager configManager;
- private final KubernetesClient kubernetesClient;
-
private final Set<FlinkResourceValidator> validators;
private final Reconciler<FlinkSessionJob> reconciler;
private final Observer<FlinkSessionJob> observer;
private final MetricManager<FlinkSessionJob> metricManager;
- private final InformerManager informerManager;
- private final Set<String> effectiveNamespaces;
-
private final StatusHelper<FlinkSessionJobStatus> statusHelper;
public FlinkSessionJobController(
FlinkConfigManager configManager,
- KubernetesClient kubernetesClient,
Set<FlinkResourceValidator> validators,
Reconciler<FlinkSessionJob> reconciler,
Observer<FlinkSessionJob> observer,
MetricManager<FlinkSessionJob> metricManager,
- StatusHelper<FlinkSessionJobStatus> statusHelper,
- InformerManager informerManager) {
+ StatusHelper<FlinkSessionJobStatus> statusHelper) {
this.configManager = configManager;
- this.kubernetesClient = kubernetesClient;
this.validators = validators;
this.reconciler = reconciler;
this.observer = observer;
this.metricManager = metricManager;
this.statusHelper = statusHelper;
- this.informerManager = informerManager;
- this.effectiveNamespaces =
configManager.getOperatorConfiguration().getWatchedNamespaces();
}
@Override
public UpdateControl<FlinkSessionJob> reconcile(
FlinkSessionJob flinkSessionJob, Context context) {
LOG.info("Starting reconciliation");
+
statusHelper.updateStatusFromCache(flinkSessionJob);
FlinkSessionJob previousJob =
ReconciliationUtils.clone(flinkSessionJob);
@@ -131,90 +115,24 @@ public class FlinkSessionJobController
}
@Override
- public Optional<FlinkSessionJob> updateErrorStatus(
- FlinkSessionJob flinkSessionJob, RetryInfo retryInfo,
RuntimeException e) {
- return ReconciliationUtils.updateErrorStatus(
- flinkSessionJob, retryInfo, e, metricManager, statusHelper);
+ public ErrorStatusUpdateControl<FlinkSessionJob> updateErrorStatus(
+ FlinkSessionJob sessionJob, Context<FlinkSessionJob> context,
Exception e) {
+ return ReconciliationUtils.toErrorStatusUpdateControl(
+ sessionJob, context.getRetryInfo(), e, metricManager,
statusHelper);
}
@Override
- public List<EventSource> prepareEventSources(
- EventSourceContext<FlinkSessionJob> eventSourceContext) {
- if (effectiveNamespaces.isEmpty()) {
- return List.of(
- createFlinkDepInformerEventSource(
-
kubernetesClient.resources(FlinkDeployment.class).inAnyNamespace(),
- OperatorUtils.ALL_NAMESPACE));
- } else {
- return effectiveNamespaces.stream()
- .map(
- name ->
- createFlinkDepInformerEventSource(
- kubernetesClient
-
.resources(FlinkDeployment.class)
- .inNamespace(name),
- name))
- .collect(Collectors.toList());
- }
- }
-
- private InformerEventSource<FlinkDeployment, FlinkSessionJob>
createFlinkDepInformerEventSource(
- FilterWatchListDeletable<FlinkDeployment,
KubernetesResourceList<FlinkDeployment>>
- filteredClient,
- String name) {
- return new InformerEventSource<>(
- filteredClient.runnableInformer(0),
- primaryResourceRetriever(),
- sessionJob ->
- new ResourceID(
- sessionJob.getSpec().getDeploymentName(),
- sessionJob.getMetadata().getNamespace()),
- false) {
- @Override
- public String name() {
- return name;
- }
- };
- }
-
- /**
- * Mapping the {@link FlinkDeployment} session cluster to {@link
FlinkSessionJob}. It leverages
- * the informer indexer.
- *
- * @return The {@link PrimaryResourcesRetriever}.
- */
- private PrimaryResourcesRetriever<FlinkDeployment>
primaryResourceRetriever() {
- return flinkDeployment -> {
- var namespace = flinkDeployment.getMetadata().getNamespace();
- var informer = informerManager.getSessionJobInformer(namespace);
-
- var sessionJobs =
- informer.getIndexer()
- .byIndex(
- OperatorUtils.CLUSTER_ID_INDEX,
- flinkDeployment.getMetadata().getName());
- var resourceIDs = new HashSet<ResourceID>();
- for (FlinkSessionJob sessionJob : sessionJobs) {
- resourceIDs.add(
- new ResourceID(
- sessionJob.getMetadata().getName(),
- sessionJob.getMetadata().getNamespace()));
- }
- LOG.debug(
- "Find the target resource {} for {} ",
- resourceIDs,
- flinkDeployment.getMetadata().getNamespace());
- return resourceIDs;
- };
+ public Map<String, EventSource> prepareEventSources(
+ EventSourceContext<FlinkSessionJob> context) {
+ return EventSourceInitializer.nameEventSources(
+
EventSourceUtils.getFlinkDeploymentInformerEventSource(context));
}
private boolean validateSessionJob(FlinkSessionJob sessionJob, Context
context) {
for (FlinkResourceValidator validator : validators) {
Optional<String> validationError =
validator.validateSessionJob(
- sessionJob,
- OperatorUtils.getSecondaryResource(
- sessionJob, context,
configManager.getOperatorConfiguration()));
+ sessionJob,
context.getSecondaryResource(FlinkDeployment.class));
if (validationError.isPresent()) {
return ReconciliationUtils.applyValidationErrorAndResetSpec(
sessionJob, validationError.get());
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
index a7d0aa3..453c41e 100644
---
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
@@ -19,22 +19,26 @@ package org.apache.flink.kubernetes.operator.informer;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
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.api.model.HasMetadata;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
+import io.javaoperatorsdk.operator.api.reconciler.Constants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.List;
+import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import static
io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_NAMESPACES_SET;
+
/** The manager of the informers. */
public class InformerManager {
private static final Logger LOG =
LoggerFactory.getLogger(InformerManager.class);
+ public static final String CLUSTER_ID_INDEX = "clusterId_index";
private final Set<String> watchedNamespaces;
private final KubernetesClient kubernetesClient;
private volatile Map<String, SharedIndexInformer<FlinkSessionJob>>
sessionJobInformers;
@@ -43,67 +47,27 @@ public class InformerManager {
public InformerManager(Set<String> watchedNamespaces, KubernetesClient
kubernetesClient) {
this.watchedNamespaces = watchedNamespaces;
this.kubernetesClient = kubernetesClient;
- LOG.info(
- "Created informer manager with watchedNamespaces: {}",
- watchedNamespaces.isEmpty()
- ? "[" + OperatorUtils.ALL_NAMESPACE + "]"
- : watchedNamespaces);
- }
-
- public SharedIndexInformer<FlinkSessionJob> getSessionJobInformer(String
namespace) {
- initSessionJobInformersWithIndexer();
- 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;
+ LOG.info("Created informer manager with watchedNamespaces: {}",
watchedNamespaces);
}
public SharedIndexInformer<FlinkDeployment> getFlinkDepInformer(String
namespace) {
initFlinkDepInformers();
var effectiveNamespace =
- watchedNamespaces.isEmpty() ? OperatorUtils.ALL_NAMESPACE :
namespace;
+ DEFAULT_NAMESPACES_SET.equals(watchedNamespaces)
+ ? Constants.WATCH_ALL_NAMESPACES
+ : namespace;
var informer = flinkDepInformers.get(effectiveNamespace);
Preconditions.checkNotNull(
informer, String.format("The informer for %s should not be
null", namespace));
return informer;
}
- private void initSessionJobInformersWithIndexer() {
- if (sessionJobInformers == null) {
- synchronized (this) {
- if (sessionJobInformers == null) {
- var runnableInformers =
- OperatorUtils.createRunnableInformer(
- FlinkSessionJob.class, watchedNamespaces,
kubernetesClient);
- for (Map.Entry<String,
SharedIndexInformer<FlinkSessionJob>> runnableInformer :
- runnableInformers.entrySet()) {
- runnableInformer
- .getValue()
- .addIndexers(
- Map.of(
- OperatorUtils.CLUSTER_ID_INDEX,
- sessionJob ->
- List.of(
- sessionJob
-
.getSpec()
-
.getDeploymentName())));
- runnableInformer.getValue().run();
- }
- this.sessionJobInformers = runnableInformers;
- LOG.info("Created session job informers for {}",
sessionJobInformers.keySet());
- }
- }
- }
- }
-
private void initFlinkDepInformers() {
if (flinkDepInformers == null) {
synchronized (this) {
if (flinkDepInformers == null) {
var runnableInformers =
- OperatorUtils.createRunnableInformer(
+ createRunnableInformer(
FlinkDeployment.class, watchedNamespaces,
kubernetesClient);
for (Map.Entry<String,
SharedIndexInformer<FlinkDeployment>> runnableInformer :
runnableInformers.entrySet()) {
@@ -117,4 +81,27 @@ public class InformerManager {
}
}
}
+
+ private static <CR extends HasMetadata>
+ Map<String, SharedIndexInformer<CR>> createRunnableInformer(
+ Class<CR> resourceClass,
+ Set<String> effectiveNamespaces,
+ KubernetesClient kubernetesClient) {
+ if (DEFAULT_NAMESPACES_SET.equals(effectiveNamespaces)) {
+ return Map.of(
+ Constants.WATCH_ALL_NAMESPACES,
+
kubernetesClient.resources(resourceClass).inAnyNamespace().runnableInformer(0));
+ } else {
+ var informers = new HashMap<String, SharedIndexInformer<CR>>();
+ for (String effectiveNamespace : effectiveNamespaces) {
+ informers.put(
+ effectiveNamespace,
+ kubernetesClient
+ .resources(resourceClass)
+ .inNamespace(effectiveNamespace)
+ .runnableInformer(0));
+ }
+ return informers;
+ }
+ }
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
index e45123c..8590f16 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
@@ -153,7 +153,7 @@ public abstract class AbstractDeploymentObserver implements
Observer<FlinkDeploy
return;
}
- Optional<Deployment> deployment = getSecondaryResource(flinkApp,
context);
+ Optional<Deployment> deployment =
context.getSecondaryResource(Deployment.class);
if (deployment.isPresent()) {
DeploymentStatus status = deployment.get().getStatus();
DeploymentSpec spec = deployment.get().getSpec();
@@ -199,14 +199,6 @@ public abstract class AbstractDeploymentObserver
implements Observer<FlinkDeploy
}
}
- private Optional<Deployment> getSecondaryResource(FlinkDeployment
flinkApp, Context context) {
- return context.getSecondaryResource(
- Deployment.class,
-
configManager.getOperatorConfiguration().getWatchedNamespaces().size() > 1
- ? flinkApp.getMetadata().getNamespace()
- : null);
- }
-
private void checkFailedCreate(DeploymentStatus status) {
List<DeploymentCondition> conditions = status.getConditions();
for (DeploymentCondition dc : conditions) {
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
index 7fa1523..d1146e9 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
@@ -18,6 +18,7 @@
package org.apache.flink.kubernetes.operator.observer.sessionjob;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+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.crd.status.JobStatus;
@@ -27,7 +28,6 @@ import
org.apache.flink.kubernetes.operator.observer.SavepointObserver;
import
org.apache.flink.kubernetes.operator.observer.context.VoidObserverContext;
import
org.apache.flink.kubernetes.operator.reconciler.sessionjob.SessionJobHelper;
import org.apache.flink.kubernetes.operator.service.FlinkService;
-import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
import org.apache.flink.kubernetes.operator.utils.StatusHelper;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.util.Preconditions;
@@ -101,9 +101,7 @@ public class SessionJobObserver implements
Observer<FlinkSessionJob> {
return;
}
- var flinkDepOpt =
- OperatorUtils.getSecondaryResource(
- flinkSessionJob, context,
configManager.getOperatorConfiguration());
+ Optional<FlinkDeployment> flinkDepOpt =
context.getSecondaryResource(FlinkDeployment.class);
var helper = new SessionJobHelper(flinkSessionJob, LOG);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
index 9b9d75f..8adfafa 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
@@ -45,6 +45,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.databind.node.TextNode;
import io.fabric8.kubernetes.client.CustomResource;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
import org.slf4j.Logger;
@@ -349,16 +350,21 @@ public class ReconciliationUtils {
* @return This always returns Empty optional currently, due to the status
update logic
*/
public static <STATUS extends CommonStatus<?>, R extends
AbstractFlinkResource<?, STATUS>>
- Optional<R> updateErrorStatus(
+ ErrorStatusUpdateControl<R> toErrorStatusUpdateControl(
R resource,
- RetryInfo retryInfo,
- RuntimeException e,
+ Optional<RetryInfo> retryInfo,
+ Exception e,
MetricManager<R> metricManager,
StatusHelper<STATUS> statusHelper) {
- LOG.warn(
- "Attempt count: {}, last attempt: {}",
- retryInfo.getAttemptCount(),
- retryInfo.isLastAttempt());
+
+ retryInfo.ifPresent(
+ r -> {
+ LOG.warn(
+ "Attempt count: {}, last attempt: {}",
+ r.getAttemptCount(),
+ r.isLastAttempt());
+ });
+
statusHelper.updateStatusFromCache(resource);
ReconciliationUtils.updateForReconciliationError(
resource,
@@ -367,6 +373,6 @@ public class ReconciliationUtils {
statusHelper.patchAndCacheStatus(resource);
// Status was updated already, no need to return anything
- return Optional.empty();
+ return ErrorStatusUpdateControl.noStatusUpdate();
}
}
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 c5140eb..7603469 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,7 +20,6 @@ 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;
@@ -36,18 +35,15 @@ 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,
- InformerManager informerManager) {
+ FlinkConfigManager configManager) {
this.kubernetesClient = kubernetesClient;
this.flinkService = flinkService;
this.configManager = configManager;
this.reconcilerMap = new ConcurrentHashMap<>();
- this.informerManager = informerManager;
}
public Reconciler<FlinkDeployment> getOrCreate(FlinkDeployment flinkApp) {
@@ -57,7 +53,7 @@ public class ReconcilerFactory {
switch (mode) {
case SESSION:
return new SessionReconciler(
- kubernetesClient, flinkService,
configManager, informerManager);
+ kubernetesClient, flinkService,
configManager);
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 246b276..18045bf 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
@@ -20,18 +20,17 @@ package
org.apache.flink.kubernetes.operator.reconciler.deployment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec;
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.EventUtils;
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;
@@ -39,6 +38,7 @@ import
io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -48,15 +48,12 @@ import java.util.stream.Collectors;
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,
- InformerManager informerManager) {
+ FlinkConfigManager configManager) {
super(kubernetesClient, flinkService, configManager);
- this.informerManager = informerManager;
}
@Override
@@ -150,11 +147,8 @@ public class SessionReconciler extends
AbstractDeploymentReconciler {
@Override
public DeleteControl cleanup(FlinkDeployment flinkApp, Context context) {
- var sessionJobs =
- informerManager
-
.getSessionJobInformer(flinkApp.getMetadata().getNamespace())
- .getIndexer()
- .byIndex(OperatorUtils.CLUSTER_ID_INDEX,
flinkApp.getMetadata().getName());
+
+ Set<FlinkSessionJob> sessionJobs =
context.getSecondaryResources(FlinkSessionJob.class);
if (!sessionJobs.isEmpty()) {
var error =
String.format(
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconciler.java
index 707dfcc..1b1eeb4 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconciler.java
@@ -31,7 +31,6 @@ import
org.apache.flink.kubernetes.operator.crd.status.Savepoint;
import org.apache.flink.kubernetes.operator.reconciler.Reconciler;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.service.FlinkService;
-import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
import org.apache.flink.util.Preconditions;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -73,8 +72,7 @@ public class FlinkSessionJobReconciler implements
Reconciler<FlinkSessionJob> {
.deserializeLastReconciledSpec();
Optional<FlinkDeployment> flinkDepOptional =
- OperatorUtils.getSecondaryResource(
- flinkSessionJob, context,
configManager.getOperatorConfiguration());
+ context.getSecondaryResource(FlinkDeployment.class);
// if session cluster is not ready, we can't do reconcile for the job.
if (!helper.sessionClusterReady(flinkDepOptional)) {
@@ -137,8 +135,7 @@ public class FlinkSessionJobReconciler implements
Reconciler<FlinkSessionJob> {
@Override
public DeleteControl cleanup(FlinkSessionJob sessionJob, Context context) {
Optional<FlinkDeployment> flinkDepOptional =
- OperatorUtils.getSecondaryResource(
- sessionJob, context,
configManager.getOperatorConfiguration());
+ context.getSecondaryResource(FlinkDeployment.class);
if (flinkDepOptional.isPresent()) {
String jobID = sessionJob.getStatus().getJobStatus().getJobId();
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EnvUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EnvUtils.java
index b82bb43..ed2c86f 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EnvUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EnvUtils.java
@@ -30,8 +30,12 @@ import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
+import java.util.Arrays;
+import java.util.HashSet;
import java.util.Properties;
+import java.util.Set;
+import static
io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_NAMESPACES_SET;
import static org.apache.flink.runtime.util.EnvironmentInformation.UNKNOWN;
import static
org.apache.flink.runtime.util.EnvironmentInformation.UNKNOWN_COMMIT_ID_ABBREV;
import static
org.apache.flink.runtime.util.EnvironmentInformation.getJvmStartupOptionsArray;
@@ -201,4 +205,16 @@ public class EnvUtils {
}
return value;
}
+
+ private static final String NAMESPACES_SPLITTER_KEY = ",";
+
+ public static Set<String> getWatchedNamespaces() {
+ String watchedNamespaces = get(EnvUtils.ENV_WATCHED_NAMESPACES);
+
+ if (StringUtils.isEmpty(watchedNamespaces)) {
+ return DEFAULT_NAMESPACES_SET;
+ } else {
+ return new
HashSet<>(Arrays.asList(watchedNamespaces.split(NAMESPACES_SPLITTER_KEY)));
+ }
+ }
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java
new file mode 100644
index 0000000..2fc26b1
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java
@@ -0,0 +1,116 @@
+/*
+ * 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.utils;
+
+import
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController;
+import
org.apache.flink.kubernetes.operator.controller.FlinkSessionJobController;
+import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
+import org.apache.flink.kubernetes.utils.Constants;
+
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import
io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.Mappers;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Utility class to locate secondary resources. */
+public class EventSourceUtils {
+
+ private static final String FLINK_DEPLOYMENT_IDX =
FlinkDeploymentController.class.getName();
+ private static final String FLINK_SESSIONJOB_IDX =
FlinkSessionJobController.class.getName();
+
+ public static InformerEventSource<Deployment, FlinkDeployment>
getDeploymentInformerEventSource(
+ EventSourceContext<FlinkDeployment> context) {
+ final String labelSelector =
+ Map.of(
+ Constants.LABEL_TYPE_KEY,
+ Constants.LABEL_TYPE_NATIVE_TYPE,
+ Constants.LABEL_COMPONENT_KEY,
+ Constants.LABEL_COMPONENT_JOB_MANAGER)
+ .entrySet().stream()
+ .map(Object::toString)
+ .collect(Collectors.joining(","));
+
+ var configuration =
+ InformerConfiguration.from(Deployment.class, context)
+ .withLabelSelector(labelSelector)
+
.withSecondaryToPrimaryMapper(Mappers.fromLabel(Constants.LABEL_APP_KEY))
+ .withNamespacesInheritedFromController(context)
+ .followNamespaceChanges(true)
+ .build();
+
+ return new InformerEventSource<>(configuration, context);
+ }
+
+ public static InformerEventSource<FlinkSessionJob, FlinkDeployment>
+
getSessionJobInformerEventSource(EventSourceContext<FlinkDeployment> context) {
+
+ context.getPrimaryCache()
+ .addIndexer(
+ FLINK_DEPLOYMENT_IDX,
+ flinkDeployment ->
List.of(flinkDeployment.getMetadata().getName()));
+
+ InformerConfiguration<FlinkSessionJob> configuration =
+ InformerConfiguration.from(FlinkSessionJob.class, context)
+ .withSecondaryToPrimaryMapper(
+ sessionJob ->
+ context.getPrimaryCache()
+ .byIndex(
+ FLINK_DEPLOYMENT_IDX,
+
sessionJob.getSpec().getDeploymentName())
+ .stream()
+ .map(ResourceID::fromResource)
+ .collect(Collectors.toSet()))
+ .withNamespacesInheritedFromController(context)
+ .followNamespaceChanges(true)
+ .build();
+
+ return new InformerEventSource<>(configuration, context);
+ }
+
+ public static InformerEventSource<FlinkDeployment, FlinkSessionJob>
+
getFlinkDeploymentInformerEventSource(EventSourceContext<FlinkSessionJob>
context) {
+ context.getPrimaryCache()
+ .addIndexer(
+ FLINK_SESSIONJOB_IDX,
+ sessionJob ->
List.of(sessionJob.getSpec().getDeploymentName()));
+
+ InformerConfiguration<FlinkDeployment> configuration =
+ InformerConfiguration.from(FlinkDeployment.class, context)
+ .withSecondaryToPrimaryMapper(
+ flinkDeployment ->
+ context.getPrimaryCache()
+ .byIndex(
+ FLINK_SESSIONJOB_IDX,
+
flinkDeployment.getMetadata().getName())
+ .stream()
+ .map(ResourceID::fromResource)
+ .collect(Collectors.toSet()))
+ .withNamespacesInheritedFromController(context)
+ .followNamespaceChanges(true)
+ .build();
+
+ return new InformerEventSource<>(configuration, 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
deleted file mode 100644
index 457fa98..0000000
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/OperatorUtils.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * 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.utils;
-
-import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
-import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
-import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
-import org.apache.flink.kubernetes.utils.Constants;
-
-import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.api.model.apps.Deployment;
-import io.fabric8.kubernetes.api.model.apps.DeploymentList;
-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.processing.event.source.informer.InformerEventSource;
-import io.javaoperatorsdk.operator.processing.event.source.informer.Mappers;
-import org.apache.commons.lang3.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-
-/** Operator SDK related utility functions. */
-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) {
- return createJmDepInformerEventSource(
- kubernetesClient.apps().deployments().inNamespace(namespace),
namespace);
- }
-
- public static InformerEventSource<Deployment, HasMetadata>
createJmDepInformerEventSource(
- KubernetesClient kubernetesClient) {
- return createJmDepInformerEventSource(
- kubernetesClient.apps().deployments().inAnyNamespace(), "all");
- }
-
- private static InformerEventSource<Deployment, HasMetadata>
createJmDepInformerEventSource(
- FilterWatchListDeletable<Deployment, DeploymentList>
filteredClient, String name) {
- SharedIndexInformer<Deployment> informer =
- filteredClient
- .withLabel(Constants.LABEL_TYPE_KEY,
Constants.LABEL_TYPE_NATIVE_TYPE)
- .withLabel(
- Constants.LABEL_COMPONENT_KEY,
- Constants.LABEL_COMPONENT_JOB_MANAGER)
- .runnableInformer(0);
-
- return new InformerEventSource<>(informer,
Mappers.fromLabel(Constants.LABEL_APP_KEY)) {
- @Override
- public String name() {
- return name;
- }
- };
- }
-
- public static Set<String> getWatchedNamespaces() {
- String watchedNamespaces =
EnvUtils.get(EnvUtils.ENV_WATCHED_NAMESPACES);
-
- if (StringUtils.isEmpty(watchedNamespaces)) {
- return Collections.emptySet();
- } else {
- return new
HashSet<>(Arrays.asList(watchedNamespaces.split(NAMESPACES_SPLITTER_KEY)));
- }
- }
-
- public static Optional<FlinkDeployment> getSecondaryResource(
- FlinkSessionJob sessionJob,
- Context context,
- FlinkOperatorConfiguration operatorConfiguration) {
- var identifier =
- operatorConfiguration.getWatchedNamespaces().size() >= 1
- ? sessionJob.getMetadata().getNamespace()
- : null;
- return context.getSecondaryResource(FlinkDeployment.class, identifier);
- }
-
- public static <CR extends HasMetadata>
- Map<String, SharedIndexInformer<CR>> createRunnableInformer(
- Class<CR> resourceClass,
- Set<String> effectiveNamespaces,
- KubernetesClient kubernetesClient) {
- if (effectiveNamespaces.isEmpty()) {
- return Map.of(
- ALL_NAMESPACE,
-
kubernetesClient.resources(resourceClass).inAnyNamespace().runnableInformer(0));
- } else {
- var informers = new HashMap<String, SharedIndexInformer<CR>>();
- for (String effectiveNamespace : effectiveNamespaces) {
- informers.put(
- effectiveNamespace,
- kubernetesClient
- .resources(resourceClass)
- .inNamespace(effectiveNamespace)
- .runnableInformer(0));
- }
- return informers;
- }
- }
-}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
index 7e5147a..d3c2a8e 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
@@ -20,7 +20,7 @@ package org.apache.flink.kubernetes.operator;
import org.apache.flink.configuration.Configuration;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
-import io.javaoperatorsdk.operator.api.config.ConfigurationService;
+import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -28,33 +28,22 @@ import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
-import static java.util.Optional.empty;
-
/** @link FlinkOperator unit tests. */
public class FlinkOperatorTest {
- @Test
- public void testExecutorServiceDefaultsToMaxParallelism() {
- checkExecutorServiceThreadCount(
- empty(),
ConfigurationService.DEFAULT_RECONCILIATION_THREADS_NUMBER);
- }
-
@Test
public void
testExecutorServiceUsesReconciliationMaxParallelismFromConfig() {
checkExecutorServiceThreadCount(Optional.of(42), 42);
- }
-
- @Test
- public void
testExecutorServiceUsesMaxParallelismForMinusOneReconciliationMaxParallelism() {
- checkExecutorServiceThreadCount(Optional.of(-1), Integer.MAX_VALUE);
+ // TODO: cannot override the operator configs twice in
java-operator-sdk v3
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> checkExecutorServiceThreadCount(Optional.of(-1), 1));
}
private void checkExecutorServiceThreadCount(
Optional<Integer> parallelism, int expectedThreadCount) {
var es = getExecutorForParallelismConfig(parallelism);
-
Assertions.assertInstanceOf(ThreadPoolExecutor.class, es);
-
ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) es;
Assertions.assertEquals(expectedThreadCount,
threadPoolExecutor.getMaximumPoolSize());
}
@@ -66,8 +55,7 @@ public class FlinkOperatorTest {
operatorConfig.setInteger(
KubernetesOperatorConfigOptions.OPERATOR_RECONCILER_MAX_PARALLELISM,
p));
-
- FlinkOperator flinkOperator = new FlinkOperator(operatorConfig);
- return
flinkOperator.getOperator().getConfigurationService().getExecutorService();
+ new FlinkOperator(operatorConfig);
+ return ConfigurationServiceProvider.instance().getExecutorService();
}
}
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 38ea772..846d641 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
@@ -38,7 +38,6 @@ 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;
@@ -61,8 +60,10 @@ import io.fabric8.kubernetes.api.model.apps.DeploymentSpec;
import io.fabric8.kubernetes.api.model.apps.DeploymentStatus;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.mockwebserver.utils.ResponseProvider;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
+import
io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext;
import okhttp3.Headers;
import okhttp3.mockwebserver.RecordedRequest;
import org.junit.jupiter.api.Assertions;
@@ -71,10 +72,10 @@ 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;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
@@ -204,9 +205,24 @@ public class TestUtils {
}
@Override
- public <T> Optional<T> getSecondaryResource(Class<T> aClass,
String s) {
+ public Optional getSecondaryResource(Class aClass, String s) {
return Optional.empty();
}
+
+ @Override
+ public Set getSecondaryResources(Class expectedType) {
+ return null;
+ }
+
+ @Override
+ public ControllerConfiguration getControllerConfiguration() {
+ return null;
+ }
+
+ @Override
+ public ManagedDependentResourceContext
managedDependentResourceContext() {
+ return null;
+ }
};
}
@@ -230,9 +246,23 @@ public class TestUtils {
}
@Override
- public <T> Optional<T> getSecondaryResource(
- Class<T> expectedType, String eventSourceName) {
- return Optional.of((T) createDeployment(true));
+ public Optional getSecondaryResource(Class expectedType, String
eventSourceName) {
+ return Optional.of(createDeployment(true));
+ }
+
+ @Override
+ public Set getSecondaryResources(Class expectedType) {
+ return null;
+ }
+
+ @Override
+ public ControllerConfiguration getControllerConfiguration() {
+ return null;
+ }
+
+ @Override
+ public ManagedDependentResourceContext
managedDependentResourceContext() {
+ return null;
}
};
}
@@ -245,9 +275,23 @@ public class TestUtils {
}
@Override
- public <T> Optional<T> getSecondaryResource(
- Class<T> expectedType, String eventSourceName) {
- return Optional.of((T) createDeployment(false));
+ public Optional getSecondaryResource(Class expectedType, String
eventSourceName) {
+ return Optional.of(createDeployment(false));
+ }
+
+ @Override
+ public Set getSecondaryResources(Class expectedType) {
+ return null;
+ }
+
+ @Override
+ public ControllerConfiguration getControllerConfiguration() {
+ return null;
+ }
+
+ @Override
+ public ManagedDependentResourceContext
managedDependentResourceContext() {
+ return null;
}
};
}
@@ -265,15 +309,29 @@ public class TestUtils {
}
@Override
- public <T> Optional<T> getSecondaryResource(
- Class<T> expectedType, String eventSourceName) {
+ public Optional getSecondaryResource(Class expectedType, String
eventSourceName) {
var session = buildSessionCluster();
session.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
session.getSpec().getFlinkConfiguration().putAll(flinkDepConfig);
session.getStatus()
.getReconciliationStatus()
.serializeAndSetLastReconciledSpec(session.getSpec());
- return Optional.of((T) session);
+ return Optional.of(session);
+ }
+
+ @Override
+ public Set getSecondaryResources(Class expectedType) {
+ return null;
+ }
+
+ @Override
+ public ControllerConfiguration getControllerConfiguration() {
+ return null;
+ }
+
+ @Override
+ public ManagedDependentResourceContext
managedDependentResourceContext() {
+ return null;
}
};
}
@@ -286,12 +344,26 @@ public class TestUtils {
}
@Override
- public <T> Optional<T> getSecondaryResource(
- Class<T> expectedType, String eventSourceName) {
+ public Optional getSecondaryResource(Class expectedType, String
eventSourceName) {
var session = buildSessionCluster();
session.getStatus()
.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.MISSING);
- return Optional.of((T) session);
+ return Optional.of(session);
+ }
+
+ @Override
+ public Set getSecondaryResources(Class expectedType) {
+ return null;
+ }
+
+ @Override
+ public ControllerConfiguration getControllerConfiguration() {
+ return null;
+ }
+
+ @Override
+ public ManagedDependentResourceContext
managedDependentResourceContext() {
+ return null;
}
};
}
@@ -306,8 +378,7 @@ public class TestUtils {
}
@Override
- public <T> Optional<T> getSecondaryResource(
- Class<T> expectedType, String eventSourceName) {
+ public Optional getSecondaryResource(Class expectedType, String
eventSourceName) {
DeploymentStatus status = new DeploymentStatus();
status.setAvailableReplicas(0);
status.setReplicas(1);
@@ -326,7 +397,22 @@ public class TestUtils {
Deployment deployment = new Deployment();
deployment.setSpec(spec);
deployment.setStatus(status);
- return Optional.of((T) deployment);
+ return Optional.of(deployment);
+ }
+
+ @Override
+ public Set getSecondaryResources(Class expectedType) {
+ return null;
+ }
+
+ @Override
+ public ControllerConfiguration getControllerConfiguration() {
+ return null;
+ }
+
+ @Override
+ public ManagedDependentResourceContext
managedDependentResourceContext() {
+ return null;
}
};
}
@@ -371,11 +457,7 @@ public class TestUtils {
configManager,
kubernetesClient,
ValidatorUtils.discoverValidators(configManager),
- new ReconcilerFactory(
- kubernetesClient,
- flinkService,
- configManager,
- new InformerManager(new HashSet<>(),
kubernetesClient)),
+ new ReconcilerFactory(kubernetesClient, flinkService,
configManager),
new ObserverFactory(kubernetesClient, flinkService,
configManager, statusHelper),
new MetricManager<>(new MetricListener().getMetricGroup()),
statusHelper);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index 2659655..27904f4 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -46,8 +46,10 @@ import
org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import io.fabric8.kubernetes.api.model.ObjectMeta;
import io.fabric8.kubernetes.api.model.PodList;
import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
+import
io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedDependentResourceContext;
import javax.annotation.Nullable;
@@ -104,11 +106,26 @@ public class TestingFlinkService extends FlinkService {
}
@Override
- public <T> Optional<T> getSecondaryResource(Class<T> aClass,
String s) {
+ public Optional getSecondaryResource(Class aClass, String s) {
if (jobs.isEmpty() && sessions.isEmpty()) {
return Optional.empty();
}
- return Optional.of((T) TestUtils.createDeployment(true));
+ return Optional.of(TestUtils.createDeployment(true));
+ }
+
+ @Override
+ public Set getSecondaryResources(Class expectedType) {
+ return null;
+ }
+
+ @Override
+ public ControllerConfiguration getControllerConfiguration() {
+ return null;
+ }
+
+ @Override
+ public ManagedDependentResourceContext
managedDependentResourceContext() {
+ return null;
}
};
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/DeploymentRecoveryTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/DeploymentRecoveryTest.java
index 60c6fa3..372aa75 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/DeploymentRecoveryTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/DeploymentRecoveryTest.java
@@ -66,7 +66,8 @@ public class DeploymentRecoveryTest {
@ParameterizedTest
@MethodSource("applicationTestParams")
- public void verifyApplicationJmRecovery(FlinkVersion flinkVersion,
UpgradeMode upgradeMode) {
+ public void verifyApplicationJmRecovery(FlinkVersion flinkVersion,
UpgradeMode upgradeMode)
+ throws Exception {
FlinkDeployment appCluster =
TestUtils.buildApplicationCluster(flinkVersion);
appCluster.getSpec().getJob().setUpgradeMode(upgradeMode);
@@ -140,7 +141,7 @@ public class DeploymentRecoveryTest {
@ParameterizedTest
@EnumSource(FlinkVersion.class)
- public void verifySessionJmRecovery(FlinkVersion flinkVersion) {
+ public void verifySessionJmRecovery(FlinkVersion flinkVersion) throws
Exception {
FlinkDeployment appCluster =
TestUtils.buildSessionCluster(flinkVersion);
testController.reconcile(appCluster, context);
testController.reconcile(appCluster, context);
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 98c4337..cab534a 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
@@ -42,7 +42,6 @@ import
io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
-import io.javaoperatorsdk.operator.processing.event.source.EventSource;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -51,12 +50,9 @@ import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.Set;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -91,7 +87,7 @@ public class FlinkDeploymentControllerTest {
@ParameterizedTest
@EnumSource(FlinkVersion.class)
- public void verifyBasicReconcileLoop(FlinkVersion flinkVersion) {
+ public void verifyBasicReconcileLoop(FlinkVersion flinkVersion) throws
Exception {
FlinkDeployment appCluster =
TestUtils.buildApplicationCluster(flinkVersion);
UpdateControl<FlinkDeployment> updateControl;
@@ -276,7 +272,7 @@ public class FlinkDeploymentControllerTest {
@ParameterizedTest
@EnumSource(FlinkVersion.class)
- public void verifyUpgradeFromSavepoint(FlinkVersion flinkVersion) {
+ public void verifyUpgradeFromSavepoint(FlinkVersion flinkVersion) throws
Exception {
FlinkDeployment appCluster =
TestUtils.buildApplicationCluster(flinkVersion);
appCluster.getSpec().getJob().setUpgradeMode(UpgradeMode.SAVEPOINT);
appCluster.getSpec().getJob().setInitialSavepointPath("s0");
@@ -365,7 +361,7 @@ public class FlinkDeploymentControllerTest {
@ParameterizedTest
@EnumSource(FlinkVersion.class)
- public void verifyStatelessUpgrade(FlinkVersion flinkVersion) {
+ public void verifyStatelessUpgrade(FlinkVersion flinkVersion) throws
Exception {
FlinkDeployment appCluster =
TestUtils.buildApplicationCluster(flinkVersion);
appCluster.getSpec().getJob().setUpgradeMode(UpgradeMode.STATELESS);
appCluster.getSpec().getJob().setInitialSavepointPath("s0");
@@ -453,21 +449,21 @@ public class FlinkDeploymentControllerTest {
@ParameterizedTest
@EnumSource(FlinkVersion.class)
- public void testUpgradeNotReadyClusterSession(FlinkVersion flinkVersion) {
+ public void testUpgradeNotReadyClusterSession(FlinkVersion flinkVersion)
throws Exception {
testUpgradeNotReadyCluster(TestUtils.buildSessionCluster(flinkVersion));
}
@ParameterizedTest
@MethodSource("applicationTestParams")
public void testUpgradeNotReadyClusterApplication(
- FlinkVersion flinkVersion, UpgradeMode upgradeMode) {
+ FlinkVersion flinkVersion, UpgradeMode upgradeMode) throws
Exception {
var appCluster = TestUtils.buildApplicationCluster(flinkVersion);
appCluster.getSpec().getJob().setUpgradeMode(upgradeMode);
testUpgradeNotReadyCluster(appCluster);
}
@Test
- public void verifyReconcileWithBadConfig() {
+ public void verifyReconcileWithBadConfig() throws Exception {
FlinkDeployment appCluster = TestUtils.buildApplicationCluster();
UpdateControl<FlinkDeployment> updateControl;
@@ -507,7 +503,7 @@ public class FlinkDeploymentControllerTest {
}
@Test
- public void verifyReconcileWithAChangedOperatorMode() {
+ public void verifyReconcileWithAChangedOperatorMode() throws Exception {
FlinkDeployment appCluster = TestUtils.buildApplicationCluster();
UpdateControl<FlinkDeployment> updateControl;
@@ -545,7 +541,7 @@ public class FlinkDeploymentControllerTest {
assertEquals(expectedJobStatus.getJobState().toString(),
jobStatus.getState());
}
- private void testUpgradeNotReadyCluster(FlinkDeployment appCluster) {
+ private void testUpgradeNotReadyCluster(FlinkDeployment appCluster) throws
Exception {
flinkService.clear();
testController.reconcile(appCluster, context);
assertEquals(
@@ -652,25 +648,7 @@ public class FlinkDeploymentControllerTest {
}
@Test
- public void testPrepareEventSource() {
- // Test watch all
- 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.setEffectiveNamespaces(namespaces);
- eventSources = testController.prepareEventSources(null);
- assertEquals(3, eventSources.size());
- assertEquals(
- namespaces,
-
eventSources.stream().map(EventSource::name).collect(Collectors.toSet()));
- }
-
- @Test
- public void testSuccessfulObservationShouldClearErrors() {
+ public void testSuccessfulObservationShouldClearErrors() throws Exception {
final String crashLoopMessage = "deploy errors";
flinkService.setJmPodList(TestUtils.createFailedPodList(crashLoopMessage));
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/informer/InformerManagerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/informer/InformerManagerTest.java
index 09715f1..08fcddd 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/informer/InformerManagerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/informer/InformerManagerTest.java
@@ -17,17 +17,16 @@
package org.apache.flink.kubernetes.operator.informer;
-import org.apache.flink.kubernetes.operator.utils.OperatorUtils;
-
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import java.util.HashSet;
import java.util.Set;
+import static
io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_NAMESPACES_SET;
+
/** Test for {@link InformerManager}. */
@EnableKubernetesMockClient(crud = true)
public class InformerManagerTest {
@@ -37,14 +36,10 @@ public class InformerManagerTest {
@Test
public void testNamespacedInformerCreated() {
- var informerManager = new InformerManager(new HashSet<>(),
kubernetesClient);
- Assertions.assertNotNull(
-
informerManager.getSessionJobInformer(OperatorUtils.ALL_NAMESPACE));
-
Assertions.assertNotNull(informerManager.getFlinkDepInformer(OperatorUtils.ALL_NAMESPACE));
+ var informerManager = new InformerManager(DEFAULT_NAMESPACES_SET,
kubernetesClient);
+ Assertions.assertNotNull(informerManager.getFlinkDepInformer("ns1"));
informerManager = new InformerManager(Set.of("ns1", "ns2"),
kubernetesClient);
- Assertions.assertNotNull(informerManager.getSessionJobInformer("ns1"));
- Assertions.assertNotNull(informerManager.getSessionJobInformer("ns2"));
Assertions.assertNotNull(informerManager.getFlinkDepInformer("ns1"));
Assertions.assertNotNull(informerManager.getFlinkDepInformer("ns2"));
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
index 1f73ab8..3d96fc2 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserverTest.java
@@ -23,20 +23,10 @@ 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.crd.status.JobManagerDeploymentStatus;
-import org.apache.flink.kubernetes.operator.observer.Observer;
-import org.apache.flink.kubernetes.operator.service.FlinkService;
-import io.fabric8.kubernetes.api.model.apps.Deployment;
-import io.fabric8.kubernetes.api.model.apps.DeploymentSpec;
-import io.fabric8.kubernetes.api.model.apps.DeploymentStatus;
import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
import org.junit.jupiter.api.Test;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
-
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -96,77 +86,4 @@ public class SessionObserverTest {
JobManagerDeploymentStatus.READY,
deployment.getStatus().getJobManagerDeploymentStatus());
}
-
- @Test
- public void testWatchMultipleNamespaces() {
- FlinkService flinkService = new TestingFlinkService();
- FlinkDeployment deployment = TestUtils.buildSessionCluster();
- deployment
- .getStatus()
- .getReconciliationStatus()
- .serializeAndSetLastReconciledSpec(deployment.getSpec());
-
- Deployment k8sDeployment = new Deployment();
- k8sDeployment.setSpec(new DeploymentSpec());
- k8sDeployment.setStatus(new DeploymentStatus());
-
- AtomicInteger secondaryResourceAccessed = new AtomicInteger(0);
- Observer observer = new SessionObserver(null, flinkService,
configManager);
- observer.observe(
- deployment,
- new Context() {
- @Override
- public Optional<RetryInfo> getRetryInfo() {
- return Optional.empty();
- }
-
- @Override
- public <T> Optional<T> getSecondaryResource(Class<T>
aClass, String s) {
- assertNull(s);
- secondaryResourceAccessed.addAndGet(1);
- return Optional.of((T) k8sDeployment);
- }
- });
-
- assertEquals(1, secondaryResourceAccessed.get());
-
-
configManager.setWatchedNamespaces(Set.of(deployment.getMetadata().getNamespace()));
- observer.observe(
- deployment,
- new Context() {
- @Override
- public Optional<RetryInfo> getRetryInfo() {
- return Optional.empty();
- }
-
- @Override
- public <T> Optional<T> getSecondaryResource(Class<T>
aClass, String s) {
- assertNull(s);
- secondaryResourceAccessed.addAndGet(1);
- return Optional.of((T) k8sDeployment);
- }
- });
-
- assertEquals(2, secondaryResourceAccessed.get());
-
-
configManager.setWatchedNamespaces(Set.of(deployment.getMetadata().getNamespace(),
"ns"));
- observer.observe(
- deployment,
- new Context() {
- @Override
- public Optional<RetryInfo> getRetryInfo() {
- return Optional.empty();
- }
-
- @Override
- public <T> Optional<T> getSecondaryResource(Class<T>
aClass, String s) {
- assertEquals(deployment.getMetadata().getNamespace(),
s);
- secondaryResourceAccessed.addAndGet(1);
- return Optional.of((T) k8sDeployment);
- }
- });
-
- configManager.setWatchedNamespaces(null);
- assertEquals(3, secondaryResourceAccessed.get());
- }
}
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 1c5ea67..0f7cfbf 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,12 +22,10 @@ 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;
@@ -38,7 +36,6 @@ 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 {
@@ -53,8 +50,7 @@ public class SessionReconcilerTest {
}
};
- SessionReconciler reconciler =
- new SessionReconciler(null, flinkService, configManager,
informerManager);
+ SessionReconciler reconciler = new SessionReconciler(null,
flinkService, configManager);
FlinkDeployment deployment = TestUtils.buildSessionCluster();
reconciler.reconcile(deployment, context);
assertEquals(1, count.get());
diff --git a/pom.xml b/pom.xml
index db2f344..052074e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -66,8 +66,9 @@ under the License.
<maven-resources-plugin.version>3.2.0</maven-resources-plugin.version>
<git-commit-id-maven-plugin.version>5.0.0</git-commit-id-maven-plugin.version>
- <operator.sdk.version>2.1.4</operator.sdk.version>
+ <operator.sdk.version>3.0.2</operator.sdk.version>
<operator.sdk.admission-controller.version>0.2.0</operator.sdk.admission-controller.version>
+
<fabric8.version>5.12.2</fabric8.version>
<lombok.version>1.18.22</lombok.version>
