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 2023/09/04 10:44:13 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-33008] Refactor to use KubernetesClient from JOSDK Context
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 3f537b83 [FLINK-33008] Refactor to use KubernetesClient from JOSDK Context
3f537b83 is described below
commit 3f537b839994ec726dd72411ecc3a0475a6ab9e4
Author: darenwkt <da...@amazon.co.uk>
AuthorDate: Mon Sep 4 11:00:56 2023 +0100
[FLINK-33008] Refactor to use KubernetesClient from JOSDK Context
---
.../operator/autoscaler/AutoscalerInfoManager.java | 35 +++---
.../operator/autoscaler/JobAutoScalerImpl.java | 28 +++--
.../autoscaler/JobAutoscalerFactoryImpl.java | 4 +-
.../operator/autoscaler/JobVertexScaler.java | 18 ++-
.../operator/autoscaler/ScalingExecutor.java | 15 ++-
.../autoscaler/AutoscalerInfoManagerTest.java | 54 +++++----
.../operator/autoscaler/AutoscalerTestUtils.java | 2 +-
.../autoscaler/BacklogBasedScalingTest.java | 13 +-
.../operator/autoscaler/JobAutoScalerImplTest.java | 18 ++-
.../operator/autoscaler/JobVertexScalerTest.java | 133 ++++++++++++++++-----
.../MetricsCollectionAndEvaluationTest.java | 11 +-
.../autoscaler/RecommendedParallelismTest.java | 12 +-
.../operator/autoscaler/ScalingExecutorTest.java | 17 ++-
.../flink/kubernetes/operator/FlinkOperator.java | 12 +-
.../controller/FlinkDeploymentController.java | 24 ++--
.../operator/controller/FlinkResourceContext.java | 10 ++
.../controller/FlinkSessionJobController.java | 18 +--
.../operator/health/CanaryResourceManager.java | 18 +--
.../observer/AbstractFlinkResourceObserver.java | 3 +-
.../operator/observer/JobStatusObserver.java | 3 +-
.../operator/observer/SnapshotObserver.java | 6 +-
.../AbstractFlinkDeploymentObserver.java | 3 +-
.../observer/deployment/ApplicationObserver.java | 3 +-
.../sessionjob/FlinkSessionJobObserver.java | 3 +-
.../operator/reconciler/ReconciliationUtils.java | 2 +-
.../AbstractFlinkResourceReconciler.java | 22 ++--
.../deployment/AbstractJobReconciler.java | 9 +-
.../deployment/ApplicationReconciler.java | 25 ++--
.../deployment/JobAutoScalerFactory.java | 4 +-
.../deployment/NoopJobAutoscalerFactory.java | 4 +-
.../reconciler/deployment/ReconcilerFactory.java | 13 +-
.../reconciler/deployment/SessionReconciler.java | 12 +-
.../sessionjob/SessionJobReconciler.java | 4 +-
.../service/FlinkResourceContextFactory.java | 8 +-
.../operator/service/NativeFlinkService.java | 3 +-
.../kubernetes/operator/utils/EventRecorder.java | 35 +++---
.../kubernetes/operator/utils/SnapshotUtils.java | 11 +-
.../kubernetes/operator/utils/StatusRecorder.java | 15 +--
.../kubernetes/operator/OperatorTestBase.java | 8 +-
.../flink/kubernetes/operator/TestUtils.java | 49 ++++++--
.../TestingFlinkResourceContextFactory.java | 4 +-
.../kubernetes/operator/TestingFlinkService.java | 5 +
.../kubernetes/operator/TestingStatusRecorder.java | 5 +-
.../controller/DeploymentRecoveryTest.java | 9 +-
.../controller/FailedDeploymentRestartTest.java | 3 +-
.../controller/FlinkDeploymentControllerTest.java | 21 ++--
.../controller/FlinkSessionJobControllerTest.java | 6 +-
.../operator/controller/RollbackTest.java | 4 +-
.../TestingFlinkDeploymentController.java | 14 +--
.../TestingFlinkSessionJobController.java | 15 +--
.../controller/UnhealthyDeploymentRestartTest.java | 3 +-
.../operator/health/HealthProbeTest.java | 14 +--
.../listener/FlinkResourceListenerTest.java | 10 +-
.../deployment/ApplicationObserverTest.java | 9 +-
.../observer/deployment/SessionObserverTest.java | 4 +-
.../sessionjob/FlinkSessionJobObserverTest.java | 20 ++--
.../deployment/ApplicationReconcilerTest.java | 35 ++----
.../ApplicationReconcilerUpgradeModeTest.java | 5 +-
.../deployment/SessionReconcilerTest.java | 3 +-
.../sessionjob/SessionJobReconcilerTest.java | 82 ++++++++-----
.../operator/service/NativeFlinkServiceTest.java | 6 +-
.../service/StandaloneFlinkServiceTest.java | 3 -
.../operator/utils/StatusRecorderTest.java | 8 +-
63 files changed, 570 insertions(+), 405 deletions(-)
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerInfoManager.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerInfoManager.java
index a3fb551a..8602743f 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerInfoManager.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerInfoManager.java
@@ -42,13 +42,11 @@ public class AutoscalerInfoManager {
private final ConcurrentHashMap<ResourceID, Optional<AutoScalerInfo>> cache =
new ConcurrentHashMap<>();
- private final KubernetesClient kubernetesClient;
- public AutoscalerInfoManager(KubernetesClient kubernetesClient) {
- this.kubernetesClient = kubernetesClient;
- }
+ public AutoscalerInfoManager() {}
- public AutoScalerInfo getOrCreateInfo(AbstractFlinkResource<?, ?> cr) {
+ public AutoScalerInfo getOrCreateInfo(
+ AbstractFlinkResource<?, ?> cr, KubernetesClient kubernetesClient) {
return cache.compute(
ResourceID.fromResource(cr),
(id, infOpt) -> {
@@ -58,12 +56,13 @@ public class AutoscalerInfoManager {
return infOpt;
}
// Otherwise get or create
- return Optional.of(getOrCreateInternal(cr));
+ return Optional.of(getOrCreateInternal(cr, kubernetesClient));
})
.get();
}
- public Optional<AutoScalerInfo> getInfo(AbstractFlinkResource<?, ?> cr) {
+ public Optional<AutoScalerInfo> getInfo(
+ AbstractFlinkResource<?, ?> cr, KubernetesClient kubernetesClient) {
return cache.compute(
ResourceID.fromResource(cr),
(id, infOpt) -> {
@@ -73,7 +72,9 @@ public class AutoscalerInfoManager {
}
// Otherwise get
- return getScalingInfoConfigMap(createCmObjectMeta(ResourceID.fromResource(cr)))
+ return getScalingInfoConfigMap(
+ createCmObjectMeta(ResourceID.fromResource(cr)),
+ kubernetesClient)
.map(AutoScalerInfo::new);
});
}
@@ -84,18 +85,23 @@ public class AutoscalerInfoManager {
}
@VisibleForTesting
- protected Optional<ConfigMap> getInfoFromKubernetes(AbstractFlinkResource<?, ?> cr) {
- return getScalingInfoConfigMap(createCmObjectMeta(ResourceID.fromResource(cr)));
+ protected Optional<ConfigMap> getInfoFromKubernetes(
+ AbstractFlinkResource<?, ?> cr, KubernetesClient kubernetesClient) {
+ return getScalingInfoConfigMap(
+ createCmObjectMeta(ResourceID.fromResource(cr)), kubernetesClient);
}
- private AutoScalerInfo getOrCreateInternal(HasMetadata cr) {
+ private AutoScalerInfo getOrCreateInternal(HasMetadata cr, KubernetesClient kubernetesClient) {
var meta = createCmObjectMeta(ResourceID.fromResource(cr));
- var info = getScalingInfoConfigMap(meta).orElseGet(() -> createConfigMap(cr, meta));
+ var info =
+ getScalingInfoConfigMap(meta, kubernetesClient)
+ .orElseGet(() -> createConfigMap(cr, meta, kubernetesClient));
return new AutoScalerInfo(info);
}
- private ConfigMap createConfigMap(HasMetadata cr, ObjectMeta meta) {
+ private ConfigMap createConfigMap(
+ HasMetadata cr, ObjectMeta meta, KubernetesClient kubernetesClient) {
LOG.info("Creating scaling info config map");
var cm = new ConfigMap();
cm.setMetadata(meta);
@@ -117,7 +123,8 @@ public class AutoscalerInfoManager {
return objectMeta;
}
- private Optional<ConfigMap> getScalingInfoConfigMap(ObjectMeta objectMeta) {
+ private Optional<ConfigMap> getScalingInfoConfigMap(
+ ObjectMeta objectMeta, KubernetesClient kubernetesClient) {
return Optional.ofNullable(
kubernetesClient
.configMaps()
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImpl.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImpl.java
index af9a8239..0b634d32 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImpl.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImpl.java
@@ -28,7 +28,6 @@ import org.apache.flink.kubernetes.operator.reconciler.deployment.JobAutoScaler;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.runtime.jobgraph.JobVertexID;
-import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,7 +44,6 @@ public class JobAutoScalerImpl implements JobAutoScaler {
private static final Logger LOG = LoggerFactory.getLogger(JobAutoScalerImpl.class);
- private final KubernetesClient kubernetesClient;
private final ScalingMetricCollector metricsCollector;
private final ScalingMetricEvaluator evaluator;
private final ScalingExecutor scalingExecutor;
@@ -61,17 +59,15 @@ public class JobAutoScalerImpl implements JobAutoScaler {
@VisibleForTesting final AutoscalerInfoManager infoManager;
public JobAutoScalerImpl(
- KubernetesClient kubernetesClient,
ScalingMetricCollector metricsCollector,
ScalingMetricEvaluator evaluator,
ScalingExecutor scalingExecutor,
EventRecorder eventRecorder) {
- this.kubernetesClient = kubernetesClient;
this.metricsCollector = metricsCollector;
this.evaluator = evaluator;
this.scalingExecutor = scalingExecutor;
this.eventRecorder = eventRecorder;
- this.infoManager = new AutoscalerInfoManager(kubernetesClient);
+ this.infoManager = new AutoscalerInfoManager();
}
@Override
@@ -89,13 +85,13 @@ public class JobAutoScalerImpl implements JobAutoScaler {
public Map<String, String> getParallelismOverrides(FlinkResourceContext<?> ctx) {
var conf = ctx.getObserveConfig();
try {
- var infoOpt = infoManager.getInfo(ctx.getResource());
+ var infoOpt = infoManager.getInfo(ctx.getResource(), ctx.getKubernetesClient());
if (infoOpt.isPresent()) {
var info = infoOpt.get();
// If autoscaler was disabled need to delete the overrides
if (!conf.getBoolean(AUTOSCALER_ENABLED) && !info.getCurrentOverrides().isEmpty()) {
info.removeCurrentOverrides();
- info.replaceInKubernetes(kubernetesClient);
+ info.replaceInKubernetes(ctx.getKubernetesClient());
} else {
return info.getCurrentOverrides();
}
@@ -130,14 +126,14 @@ public class JobAutoScalerImpl implements JobAutoScaler {
return false;
}
- var autoScalerInfo = infoManager.getOrCreateInfo(resource);
+ var autoScalerInfo = infoManager.getOrCreateInfo(resource, ctx.getKubernetesClient());
var collectedMetrics =
metricsCollector.updateMetrics(
resource, autoScalerInfo, ctx.getFlinkService(), conf);
if (collectedMetrics.getMetricHistory().isEmpty()) {
- autoScalerInfo.replaceInKubernetes(kubernetesClient);
+ autoScalerInfo.replaceInKubernetes(ctx.getKubernetesClient());
return false;
}
LOG.debug("Collected metrics: {}", collectedMetrics);
@@ -154,12 +150,17 @@ public class JobAutoScalerImpl implements JobAutoScaler {
if (!collectedMetrics.isFullyCollected()) {
// We have done an upfront evaluation, but we are not ready for scaling.
resetRecommendedParallelism(evaluatedMetrics);
- autoScalerInfo.replaceInKubernetes(kubernetesClient);
+ autoScalerInfo.replaceInKubernetes(ctx.getKubernetesClient());
return false;
}
var specAdjusted =
- scalingExecutor.scaleResource(resource, autoScalerInfo, conf, evaluatedMetrics);
+ scalingExecutor.scaleResource(
+ resource,
+ autoScalerInfo,
+ conf,
+ evaluatedMetrics,
+ ctx.getKubernetesClient());
if (specAdjusted) {
autoscalerMetrics.numScalings.inc();
@@ -167,7 +168,7 @@ public class JobAutoScalerImpl implements JobAutoScaler {
autoscalerMetrics.numBalanced.inc();
}
- autoScalerInfo.replaceInKubernetes(kubernetesClient);
+ autoScalerInfo.replaceInKubernetes(ctx.getKubernetesClient());
return specAdjusted;
} catch (Throwable e) {
LOG.error("Error while scaling resource", e);
@@ -177,7 +178,8 @@ public class JobAutoScalerImpl implements JobAutoScaler {
EventRecorder.Type.Warning,
EventRecorder.Reason.AutoscalerError,
EventRecorder.Component.Operator,
- e.getMessage());
+ e.getMessage(),
+ ctx.getKubernetesClient());
return false;
}
}
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoscalerFactoryImpl.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoscalerFactoryImpl.java
index 277796e5..f6a7afc8 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoscalerFactoryImpl.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoscalerFactoryImpl.java
@@ -22,7 +22,6 @@ import org.apache.flink.kubernetes.operator.reconciler.deployment.JobAutoScalerF
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import com.google.auto.service.AutoService;
-import io.fabric8.kubernetes.client.KubernetesClient;
/**
* Factory for loading JobAutoScalerImpl included in this module. This class will be dynamically
@@ -31,9 +30,8 @@ import io.fabric8.kubernetes.client.KubernetesClient;
@AutoService(JobAutoScalerFactory.class)
public class JobAutoscalerFactoryImpl implements JobAutoScalerFactory {
@Override
- public JobAutoScaler create(KubernetesClient kubernetesClient, EventRecorder eventRecorder) {
+ public JobAutoScaler create(EventRecorder eventRecorder) {
return new JobAutoScalerImpl(
- kubernetesClient,
new RestApiMetricsCollector(),
new ScalingMetricEvaluator(),
new ScalingExecutor(eventRecorder),
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobVertexScaler.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobVertexScaler.java
index 497bb270..fec126af 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobVertexScaler.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/JobVertexScaler.java
@@ -28,6 +28,7 @@ import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.util.Preconditions;
+import io.fabric8.kubernetes.client.KubernetesClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,7 +71,8 @@ public class JobVertexScaler {
Configuration conf,
JobVertexID vertex,
Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
- SortedMap<Instant, ScalingSummary> history) {
+ SortedMap<Instant, ScalingSummary> history,
+ KubernetesClient client) {
var currentParallelism = (int) evaluatedMetrics.get(PARALLELISM).getCurrent();
double averageTrueProcessingRate = evaluatedMetrics.get(TRUE_PROCESSING_RATE).getAverage();
@@ -131,7 +133,8 @@ public class JobVertexScaler {
evaluatedMetrics,
history,
currentParallelism,
- newParallelism)) {
+ newParallelism,
+ client)) {
return currentParallelism;
}
@@ -149,7 +152,8 @@ public class JobVertexScaler {
Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
SortedMap<Instant, ScalingSummary> history,
int currentParallelism,
- int newParallelism) {
+ int newParallelism,
+ KubernetesClient client) {
// If we don't have past scaling actions for this vertex, there is nothing to do
if (history.isEmpty()) {
@@ -163,7 +167,7 @@ public class JobVertexScaler {
if (currentParallelism == lastSummary.getNewParallelism() && lastSummary.isScaledUp()) {
if (scaledUp) {
return detectIneffectiveScaleUp(
- resource, vertex, conf, evaluatedMetrics, lastSummary);
+ resource, vertex, conf, evaluatedMetrics, lastSummary, client);
} else {
return detectImmediateScaleDownAfterScaleUp(vertex, conf, lastScalingTs);
}
@@ -190,7 +194,8 @@ public class JobVertexScaler {
JobVertexID vertex,
Configuration conf,
Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
- ScalingSummary lastSummary) {
+ ScalingSummary lastSummary,
+ KubernetesClient client) {
double lastProcRate = lastSummary.getMetrics().get(TRUE_PROCESSING_RATE).getAverage();
double lastExpectedProcRate =
@@ -218,7 +223,8 @@ public class JobVertexScaler {
EventRecorder.Type.Normal,
EventRecorder.Reason.IneffectiveScaling,
EventRecorder.Component.Operator,
- message);
+ message,
+ client);
if (conf.get(AutoScalerOptions.SCALING_EFFECTIVENESS_DETECTION_ENABLED)) {
LOG.warn(
diff --git a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java
index 8ede4d99..1d6a9cd0 100644
--- a/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java
+++ b/flink-kubernetes-operator-autoscaler/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutor.java
@@ -27,6 +27,7 @@ import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.util.Preconditions;
+import io.fabric8.kubernetes.client.KubernetesClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -71,12 +72,13 @@ public class ScalingExecutor {
AbstractFlinkResource<?, ?> resource,
AutoScalerInfo scalingInformation,
Configuration conf,
- Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedMetrics) {
+ Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedMetrics,
+ KubernetesClient client) {
var now = Instant.now();
var scalingHistory = scalingInformation.getScalingHistory(now, conf);
var scalingSummaries =
- computeScalingSummary(resource, conf, evaluatedMetrics, scalingHistory);
+ computeScalingSummary(resource, conf, evaluatedMetrics, scalingHistory, client);
if (scalingSummaries.isEmpty()) {
LOG.info("All job vertices are currently running at their target parallelism.");
@@ -98,7 +100,8 @@ public class ScalingExecutor {
EventRecorder.Reason.ScalingReport,
EventRecorder.Component.Operator,
scalingReport,
- "ScalingExecutor");
+ "ScalingExecutor",
+ client);
if (!scalingEnabled) {
return false;
@@ -183,7 +186,8 @@ public class ScalingExecutor {
AbstractFlinkResource<?, ?> resource,
Configuration conf,
Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedMetrics,
- Map<JobVertexID, SortedMap<Instant, ScalingSummary>> scalingHistory) {
+ Map<JobVertexID, SortedMap<Instant, ScalingSummary>> scalingHistory,
+ KubernetesClient client) {
var out = new HashMap<JobVertexID, ScalingSummary>();
var excludeVertexIdList = conf.get(AutoScalerOptions.VERTEX_EXCLUDE_IDS);
@@ -203,7 +207,8 @@ public class ScalingExecutor {
v,
metrics,
scalingHistory.getOrDefault(
- v, Collections.emptySortedMap()));
+ v, Collections.emptySortedMap()),
+ client);
if (currentParallelism != newParallelism) {
out.put(
v,
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerInfoManagerTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerInfoManagerTest.java
index f2eb1058..4c51d05a 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerInfoManagerTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerInfoManagerTest.java
@@ -47,46 +47,54 @@ public class AutoscalerInfoManagerTest {
var cr2 = TestUtils.buildApplicationCluster();
cr2.getMetadata().setName("cr2");
- var manager = new AutoscalerInfoManager(kubernetesClient);
+ var manager = new AutoscalerInfoManager();
assertEquals(0, mockWebServer.getRequestCount());
- assertTrue(manager.getInfo(cr1).isEmpty());
+ assertTrue(manager.getInfo(cr1, kubernetesClient).isEmpty());
assertEquals(1, mockWebServer.getRequestCount());
// Further gets should not go to K8s
- assertTrue(manager.getInfo(cr1).isEmpty());
- assertTrue(manager.getInfo(cr1).isEmpty());
+ assertTrue(manager.getInfo(cr1, kubernetesClient).isEmpty());
+ assertTrue(manager.getInfo(cr1, kubernetesClient).isEmpty());
assertEquals(1, mockWebServer.getRequestCount());
- assertTrue(manager.getInfoFromKubernetes(cr1).isEmpty());
+ assertTrue(manager.getInfoFromKubernetes(cr1, kubernetesClient).isEmpty());
assertEquals(2, mockWebServer.getRequestCount());
- var info1 = manager.getOrCreateInfo(cr1);
+ var info1 = manager.getOrCreateInfo(cr1, kubernetesClient);
assertEquals(4, mockWebServer.getRequestCount());
- assertTrue(info1 == manager.getOrCreateInfo(cr1));
+ assertTrue(info1 == manager.getOrCreateInfo(cr1, kubernetesClient));
assertEquals(4, mockWebServer.getRequestCount());
- assertFalse(manager.getInfoFromKubernetes(cr1).isEmpty());
- assertTrue(manager.getInfoFromKubernetes(cr2).isEmpty());
+ assertFalse(manager.getInfoFromKubernetes(cr1, kubernetesClient).isEmpty());
+ assertTrue(manager.getInfoFromKubernetes(cr2, kubernetesClient).isEmpty());
- assertFalse(manager.getInfo(cr1).isEmpty());
- assertTrue(manager.getInfo(cr2).isEmpty());
+ assertFalse(manager.getInfo(cr1, kubernetesClient).isEmpty());
+ assertTrue(manager.getInfo(cr2, kubernetesClient).isEmpty());
- var info2 = manager.getOrCreateInfo(cr2);
+ var info2 = manager.getOrCreateInfo(cr2, kubernetesClient);
info1.setCurrentOverrides(Map.of("a", "1"));
info1.replaceInKubernetes(kubernetesClient);
info2.setCurrentOverrides(Map.of("b", "1"));
info2.replaceInKubernetes(kubernetesClient);
- manager = new AutoscalerInfoManager(kubernetesClient);
- assertEquals(Map.of("a", "1"), manager.getInfo(cr1).get().getCurrentOverrides());
- assertEquals(Map.of("b", "1"), manager.getOrCreateInfo(cr2).getCurrentOverrides());
+ manager = new AutoscalerInfoManager();
+ assertEquals(
+ Map.of("a", "1"),
+ manager.getInfo(cr1, kubernetesClient).get().getCurrentOverrides());
+ assertEquals(
+ Map.of("b", "1"),
+ manager.getOrCreateInfo(cr2, kubernetesClient).getCurrentOverrides());
// Removing from the cache should not affect the results
manager.removeInfoFromCache(cr1);
- assertEquals(Map.of("a", "1"), manager.getInfo(cr1).get().getCurrentOverrides());
- assertEquals(Map.of("b", "1"), manager.getOrCreateInfo(cr2).getCurrentOverrides());
+ assertEquals(
+ Map.of("a", "1"),
+ manager.getInfo(cr1, kubernetesClient).get().getCurrentOverrides());
+ assertEquals(
+ Map.of("b", "1"),
+ manager.getOrCreateInfo(cr2, kubernetesClient).getCurrentOverrides());
}
@Test
@@ -94,10 +102,10 @@ public class AutoscalerInfoManagerTest {
var cr1 = TestUtils.buildApplicationCluster();
cr1.getMetadata().setName("cr1");
- var manager = new AutoscalerInfoManager(kubernetesClient);
+ var manager = new AutoscalerInfoManager();
assertEquals(0, mockWebServer.getRequestCount());
- var info1 = manager.getOrCreateInfo(cr1);
+ var info1 = manager.getOrCreateInfo(cr1, kubernetesClient);
assertTrue(info1.isValid());
assertEquals(2, mockWebServer.getRequestCount());
@@ -114,7 +122,11 @@ public class AutoscalerInfoManagerTest {
assertFalse(info1.isValid());
// Make sure we can get the new version
- assertEquals(Map.of("a", "3"), manager.getInfo(cr1).get().getCurrentOverrides());
- assertEquals(Map.of("a", "3"), manager.getOrCreateInfo(cr1).getCurrentOverrides());
+ assertEquals(
+ Map.of("a", "3"),
+ manager.getInfo(cr1, kubernetesClient).get().getCurrentOverrides());
+ assertEquals(
+ Map.of("a", "3"),
+ manager.getOrCreateInfo(cr1, kubernetesClient).getCurrentOverrides());
}
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerTestUtils.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerTestUtils.java
index 0ce502db..17265b90 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerTestUtils.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerTestUtils.java
@@ -28,6 +28,6 @@ public class AutoscalerTestUtils {
@SneakyThrows
public static AutoScalerInfo getOrCreateInfo(
AbstractFlinkResource<?, ?> cr, KubernetesClient client) {
- return new AutoscalerInfoManager(client).getOrCreateInfo(cr);
+ return new AutoscalerInfoManager().getOrCreateInfo(cr, client);
}
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/BacklogBasedScalingTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/BacklogBasedScalingTest.java
index 2c76ae7d..37b105bb 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/BacklogBasedScalingTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/BacklogBasedScalingTest.java
@@ -76,8 +76,7 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
@BeforeEach
public void setup() {
evaluator = new ScalingMetricEvaluator();
- scalingExecutor =
- new ScalingExecutor(new EventRecorder(kubernetesClient, new EventCollector()));
+ scalingExecutor = new ScalingExecutor(new EventRecorder(new EventCollector()));
app = TestUtils.buildApplicationCluster();
app.getMetadata().setGeneration(1L);
@@ -113,11 +112,10 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
autoscaler =
new JobAutoScalerImpl(
- kubernetesClient,
metricsCollector,
evaluator,
scalingExecutor,
- new EventRecorder(kubernetesClient, eventCollector));
+ new EventRecorder(eventCollector));
// Reset custom window size to default
metricsCollector.setTestMetricWindowSize(null);
@@ -428,11 +426,18 @@ public class BacklogBasedScalingTest extends OperatorTestBase {
@NotNull
private TestUtils.TestingContext<HasMetadata> createAutoscalerTestContext() {
return new TestUtils.TestingContext<>() {
+
+ @Override
public <T1> Set<T1> getSecondaryResources(Class<T1> aClass) {
return (Set)
kubernetesClient.configMaps().inAnyNamespace().list().getItems().stream()
.collect(Collectors.toSet());
}
+
+ @Override
+ public KubernetesClient getClient() {
+ return kubernetesClient;
+ }
};
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImplTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImplTest.java
index 16c24500..030093e8 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImplTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobAutoScalerImplTest.java
@@ -101,12 +101,7 @@ public class JobAutoScalerImplTest extends OperatorTestBase {
ScalingExecutor scalingExecutor = new ScalingExecutor(eventRecorder);
var autoscaler =
- new JobAutoScalerImpl(
- kubernetesClient,
- metricsCollector,
- evaluator,
- scalingExecutor,
- eventRecorder);
+ new JobAutoScalerImpl(metricsCollector, evaluator, scalingExecutor, eventRecorder);
FlinkResourceContext<FlinkDeployment> resourceContext = getResourceContext(app);
ResourceID resourceId = ResourceID.fromResource(app);
@@ -139,7 +134,7 @@ public class JobAutoScalerImplTest extends OperatorTestBase {
@Test
void testErrorReporting() {
- var autoscaler = new JobAutoScalerImpl(kubernetesClient, null, null, null, eventRecorder);
+ var autoscaler = new JobAutoScalerImpl(null, null, null, eventRecorder);
FlinkResourceContext<FlinkDeployment> resourceContext = getResourceContext(app);
ResourceID resourceId = ResourceID.fromResource(app);
@@ -154,14 +149,15 @@ public class JobAutoScalerImplTest extends OperatorTestBase {
@Test
void testParallelismOverrides() throws Exception {
- var autoscaler = new JobAutoScalerImpl(kubernetesClient, null, null, null, eventRecorder);
+ var autoscaler = new JobAutoScalerImpl(null, null, null, eventRecorder);
var ctx = getResourceContext(app);
// Initially we should return empty overrides, do not crate any CM
assertEquals(Map.of(), autoscaler.getParallelismOverrides(ctx));
- assertFalse(autoscaler.infoManager.getInfoFromKubernetes(app).isPresent());
+ assertFalse(
+ autoscaler.infoManager.getInfoFromKubernetes(app, kubernetesClient).isPresent());
- var autoscalerInfo = autoscaler.infoManager.getOrCreateInfo(app);
+ var autoscalerInfo = autoscaler.infoManager.getOrCreateInfo(app, kubernetesClient);
var v1 = new JobVertexID().toString();
var v2 = new JobVertexID().toString();
@@ -175,7 +171,7 @@ public class JobAutoScalerImplTest extends OperatorTestBase {
ctx = getResourceContext(app);
assertEquals(Map.of(), autoscaler.getParallelismOverrides(ctx));
// But not clear the autoscaler info
- assertTrue(autoscaler.infoManager.getInfoFromKubernetes(app).isPresent());
+ assertTrue(autoscaler.infoManager.getInfoFromKubernetes(app, kubernetesClient).isPresent());
int requestCount = mockWebServer.getRequestCount();
// Make sure we don't update in kubernetes once removed
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobVertexScalerTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobVertexScalerTest.java
index fb4655df..8bff55be 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobVertexScalerTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/JobVertexScalerTest.java
@@ -63,7 +63,7 @@ public class JobVertexScalerTest {
flinkDep = TestUtils.buildApplicationCluster();
kubernetesClient.resource(flinkDep).createOrReplace();
eventCollector = new EventCollector();
- vertexScaler = new JobVertexScaler(new EventRecorder(kubernetesClient, eventCollector));
+ vertexScaler = new JobVertexScaler(new EventRecorder(eventCollector));
conf = new Configuration();
conf.set(AutoScalerOptions.MAX_SCALE_DOWN_FACTOR, 1.);
conf.set(AutoScalerOptions.MAX_SCALE_UP_FACTOR, (double) Integer.MAX_VALUE);
@@ -77,68 +77,123 @@ public class JobVertexScalerTest {
assertEquals(
5,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(10, 50, 100), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(10, 50, 100),
+ Collections.emptySortedMap(),
+ kubernetesClient));
conf.set(AutoScalerOptions.TARGET_UTILIZATION, .8);
assertEquals(
8,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(10, 50, 100), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(10, 50, 100),
+ Collections.emptySortedMap(),
+ kubernetesClient));
conf.set(AutoScalerOptions.TARGET_UTILIZATION, .8);
assertEquals(
10,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(10, 80, 100), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(10, 80, 100),
+ Collections.emptySortedMap(),
+ kubernetesClient));
conf.set(AutoScalerOptions.TARGET_UTILIZATION, .8);
assertEquals(
8,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(10, 60, 100), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(10, 60, 100),
+ Collections.emptySortedMap(),
+ kubernetesClient));
assertEquals(
8,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(10, 59, 100), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(10, 59, 100),
+ Collections.emptySortedMap(),
+ kubernetesClient));
conf.set(AutoScalerOptions.TARGET_UTILIZATION, 0.5);
assertEquals(
10,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(2, 100, 40), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(2, 100, 40),
+ Collections.emptySortedMap(),
+ kubernetesClient));
conf.set(AutoScalerOptions.TARGET_UTILIZATION, 0.6);
assertEquals(
4,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(2, 100, 100), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(2, 100, 100),
+ Collections.emptySortedMap(),
+ kubernetesClient));
conf.set(AutoScalerOptions.TARGET_UTILIZATION, 1.);
conf.set(AutoScalerOptions.MAX_SCALE_DOWN_FACTOR, 0.5);
assertEquals(
5,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(10, 10, 100), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(10, 10, 100),
+ Collections.emptySortedMap(),
+ kubernetesClient));
conf.set(AutoScalerOptions.MAX_SCALE_DOWN_FACTOR, 0.6);
assertEquals(
4,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(10, 10, 100), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(10, 10, 100),
+ Collections.emptySortedMap(),
+ kubernetesClient));
conf.set(AutoScalerOptions.TARGET_UTILIZATION, 1.);
conf.set(AutoScalerOptions.MAX_SCALE_UP_FACTOR, 0.5);
assertEquals(
15,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(10, 200, 10), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(10, 200, 10),
+ Collections.emptySortedMap(),
+ kubernetesClient));
conf.set(AutoScalerOptions.MAX_SCALE_UP_FACTOR, 0.6);
assertEquals(
16,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, op, evaluated(10, 200, 10), Collections.emptySortedMap()));
+ flinkDep,
+ conf,
+ op,
+ evaluated(10, 200, 10),
+ Collections.emptySortedMap(),
+ kubernetesClient));
}
@Test
@@ -186,7 +241,8 @@ public class JobVertexScalerTest {
conf,
new JobVertexID(),
evaluated(10, 100, 500),
- Collections.emptySortedMap()));
+ Collections.emptySortedMap(),
+ kubernetesClient));
// Make sure we respect current parallelism in case it's lower
assertEquals(
@@ -196,7 +252,8 @@ public class JobVertexScalerTest {
conf,
new JobVertexID(),
evaluated(4, 100, 500),
- Collections.emptySortedMap()));
+ Collections.emptySortedMap(),
+ kubernetesClient));
}
@Test
@@ -210,7 +267,8 @@ public class JobVertexScalerTest {
conf,
new JobVertexID(),
evaluated(10, 500, 100),
- Collections.emptySortedMap()));
+ Collections.emptySortedMap(),
+ kubernetesClient));
// Make sure we respect current parallelism in case it's higher
assertEquals(
@@ -220,7 +278,8 @@ public class JobVertexScalerTest {
conf,
new JobVertexID(),
evaluated(12, 500, 100),
- Collections.emptySortedMap()));
+ Collections.emptySortedMap(),
+ kubernetesClient));
}
@Test
@@ -235,7 +294,8 @@ public class JobVertexScalerTest {
var history = new TreeMap<Instant, ScalingSummary>();
assertEquals(
10,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
history.put(clock.instant(), new ScalingSummary(5, 10, evaluated));
@@ -243,7 +303,8 @@ public class JobVertexScalerTest {
evaluated = evaluated(10, 50, 100);
assertEquals(
10,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
// Pass some time...
clock = Clock.offset(Clock.systemDefaultZone(), Duration.ofSeconds(61));
@@ -251,14 +312,16 @@ public class JobVertexScalerTest {
assertEquals(
5,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
history.put(clock.instant(), new ScalingSummary(10, 5, evaluated));
// Allow immediate scale up
evaluated = evaluated(5, 100, 50);
assertEquals(
10,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
history.put(clock.instant(), new ScalingSummary(5, 10, evaluated));
}
@@ -273,7 +336,8 @@ public class JobVertexScalerTest {
var history = new TreeMap<Instant, ScalingSummary>();
assertEquals(
10,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
assertEquals(100, evaluated.get(ScalingMetric.EXPECTED_PROCESSING_RATE).getCurrent());
history.put(Instant.now(), new ScalingSummary(5, 10, evaluated));
@@ -281,7 +345,8 @@ public class JobVertexScalerTest {
evaluated = evaluated(10, 180, 90);
assertEquals(
20,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
assertEquals(180, evaluated.get(ScalingMetric.EXPECTED_PROCESSING_RATE).getCurrent());
history.put(Instant.now(), new ScalingSummary(10, 20, evaluated));
@@ -290,27 +355,31 @@ public class JobVertexScalerTest {
evaluated = evaluated(20, 180, 94);
assertEquals(
20,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
assertFalse(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
// Still considered ineffective (less than <10%)
evaluated = evaluated(20, 180, 98);
assertEquals(
20,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
assertFalse(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
// Allow scale up if current parallelism doesnt match last (user rescaled manually)
evaluated = evaluated(10, 180, 90);
assertEquals(
20,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
// Over 10%, effective
evaluated = evaluated(20, 180, 100);
assertEquals(
36,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
assertTrue(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
// Ineffective but detection is turned off
@@ -318,7 +387,8 @@ public class JobVertexScalerTest {
evaluated = evaluated(20, 180, 90);
assertEquals(
40,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
assertTrue(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
conf.set(AutoScalerOptions.SCALING_EFFECTIVENESS_DETECTION_ENABLED, true);
@@ -326,7 +396,8 @@ public class JobVertexScalerTest {
evaluated = evaluated(20, 45, 90);
assertEquals(
10,
- vertexScaler.computeScaleTargetParallelism(flinkDep, conf, op, evaluated, history));
+ vertexScaler.computeScaleTargetParallelism(
+ flinkDep, conf, op, evaluated, history, kubernetesClient));
assertTrue(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
}
@@ -342,7 +413,7 @@ public class JobVertexScalerTest {
assertEquals(
10,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, jobVertexID, evaluated, history));
+ flinkDep, conf, jobVertexID, evaluated, history, kubernetesClient));
assertEquals(100, evaluated.get(ScalingMetric.EXPECTED_PROCESSING_RATE).getCurrent());
history.put(Instant.now(), new ScalingSummary(5, 10, evaluated));
@@ -351,7 +422,7 @@ public class JobVertexScalerTest {
assertEquals(
20,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, jobVertexID, evaluated, history));
+ flinkDep, conf, jobVertexID, evaluated, history, kubernetesClient));
assertEquals(180, evaluated.get(ScalingMetric.EXPECTED_PROCESSING_RATE).getCurrent());
history.put(Instant.now(), new ScalingSummary(10, 20, evaluated));
assertEquals(0, eventCollector.events.size());
@@ -361,7 +432,7 @@ public class JobVertexScalerTest {
assertEquals(
20,
vertexScaler.computeScaleTargetParallelism(
- flinkDep, conf, jobVertexID, evaluated, history));
+ flinkDep, conf, jobVertexID, evaluated, history, kubernetesClient));
assertFalse(evaluated.containsKey(ScalingMetric.EXPECTED_PROCESSING_RATE));
assertEquals(1, eventCollector.events.size());
var event = eventCollector.events.poll();
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/MetricsCollectionAndEvaluationTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/MetricsCollectionAndEvaluationTest.java
index 0d973b00..a01195c2 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/MetricsCollectionAndEvaluationTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/MetricsCollectionAndEvaluationTest.java
@@ -87,8 +87,7 @@ public class MetricsCollectionAndEvaluationTest {
@BeforeEach
public void setup() {
evaluator = new ScalingMetricEvaluator();
- scalingExecutor =
- new ScalingExecutor(new EventRecorder(kubernetesClient, new EventCollector()));
+ scalingExecutor = new ScalingExecutor(new EventRecorder(new EventCollector()));
service = new TestingFlinkService();
app = TestUtils.buildApplicationCluster();
@@ -176,7 +175,7 @@ public class MetricsCollectionAndEvaluationTest {
assertTrue(collectedMetrics.isFullyCollected());
var evaluation = evaluator.evaluate(conf, collectedMetrics);
- scalingExecutor.scaleResource(app, scalingInfo, conf, evaluation);
+ scalingExecutor.scaleResource(app, scalingInfo, conf, evaluation, kubernetesClient);
var scaledParallelism = ScalingExecutorTest.getScaledParallelism(scalingInfo);
assertEquals(4, scaledParallelism.size());
@@ -190,7 +189,7 @@ public class MetricsCollectionAndEvaluationTest {
conf.set(AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY, 0.);
evaluation = evaluator.evaluate(conf, collectedMetrics);
- scalingExecutor.scaleResource(app, scalingInfo, conf, evaluation);
+ scalingExecutor.scaleResource(app, scalingInfo, conf, evaluation, kubernetesClient);
scaledParallelism = ScalingExecutorTest.getScaledParallelism(scalingInfo);
assertEquals(4, scaledParallelism.get(source1));
@@ -398,7 +397,7 @@ public class MetricsCollectionAndEvaluationTest {
500.,
evaluation.get(source1).get(ScalingMetric.SCALE_UP_RATE_THRESHOLD).getCurrent());
- scalingExecutor.scaleResource(app, scalingInfo, conf, evaluation);
+ scalingExecutor.scaleResource(app, scalingInfo, conf, evaluation, kubernetesClient);
var scaledParallelism = ScalingExecutorTest.getScaledParallelism(scalingInfo);
assertEquals(1, scaledParallelism.get(source1));
}
@@ -478,7 +477,7 @@ public class MetricsCollectionAndEvaluationTest {
0.,
evaluation.get(source1).get(ScalingMetric.SCALE_UP_RATE_THRESHOLD).getCurrent());
- scalingExecutor.scaleResource(app, scalingInfo, conf, evaluation);
+ scalingExecutor.scaleResource(app, scalingInfo, conf, evaluation, kubernetesClient);
var scaledParallelism = ScalingExecutorTest.getScaledParallelism(scalingInfo);
assertEquals(1, scaledParallelism.get(source1));
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/RecommendedParallelismTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/RecommendedParallelismTest.java
index 23ee4055..b8d1528b 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/RecommendedParallelismTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/RecommendedParallelismTest.java
@@ -79,8 +79,7 @@ public class RecommendedParallelismTest extends OperatorTestBase {
@BeforeEach
public void setup() {
evaluator = new ScalingMetricEvaluator();
- scalingExecutor =
- new ScalingExecutor(new EventRecorder(kubernetesClient, new EventCollector()));
+ scalingExecutor = new ScalingExecutor(new EventRecorder(new EventCollector()));
app = TestUtils.buildApplicationCluster();
app.getMetadata().setGeneration(1L);
@@ -116,11 +115,10 @@ public class RecommendedParallelismTest extends OperatorTestBase {
autoscaler =
new JobAutoScalerImpl(
- kubernetesClient,
metricsCollector,
evaluator,
scalingExecutor,
- new EventRecorder(kubernetesClient, eventCollector));
+ new EventRecorder(eventCollector));
// Reset custom window size to default
metricsCollector.setTestMetricWindowSize(null);
@@ -275,11 +273,17 @@ public class RecommendedParallelismTest extends OperatorTestBase {
@NotNull
private TestUtils.TestingContext<HasMetadata> createAutoscalerTestContext() {
return new TestUtils.TestingContext<>() {
+ @Override
public <T1> Set<T1> getSecondaryResources(Class<T1> aClass) {
return (Set)
kubernetesClient.configMaps().inAnyNamespace().list().getItems().stream()
.collect(Collectors.toSet());
}
+
+ @Override
+ public KubernetesClient getClient() {
+ return kubernetesClient;
+ }
};
}
}
diff --git a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java
index 2191f723..f3a26525 100644
--- a/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java
+++ b/flink-kubernetes-operator-autoscaler/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/ScalingExecutorTest.java
@@ -64,8 +64,7 @@ public class ScalingExecutorTest {
@BeforeEach
public void setup() {
eventCollector = new EventCollector();
- scalingDecisionExecutor =
- new ScalingExecutor(new EventRecorder(kubernetesClient, eventCollector));
+ scalingDecisionExecutor = new ScalingExecutor(new EventRecorder(eventCollector));
conf = new Configuration();
conf.set(AutoScalerOptions.STABILIZATION_INTERVAL, Duration.ZERO);
conf.set(AutoScalerOptions.SCALING_ENABLED, true);
@@ -158,10 +157,14 @@ public class ScalingExecutorTest {
evaluated(10, 80, 100));
// filter operator should not scale
conf.set(AutoScalerOptions.VERTEX_EXCLUDE_IDS, List.of(filterOperatorHexString));
- assertFalse(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
+ assertFalse(
+ scalingDecisionExecutor.scaleResource(
+ flinkDep, scalingInfo, conf, metrics, kubernetesClient));
// filter operator should scale
conf.set(AutoScalerOptions.VERTEX_EXCLUDE_IDS, List.of());
- assertTrue(scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
+ assertTrue(
+ scalingDecisionExecutor.scaleResource(
+ flinkDep, scalingInfo, conf, metrics, kubernetesClient));
}
@ParameterizedTest
@@ -173,7 +176,8 @@ public class ScalingExecutorTest {
var scalingInfo = new AutoScalerInfo(new HashMap<>());
assertEquals(
scalingEnabled,
- scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
+ scalingDecisionExecutor.scaleResource(
+ flinkDep, scalingInfo, conf, metrics, kubernetesClient));
assertEquals(1, eventCollector.events.size());
var event = eventCollector.events.poll();
assertTrue(
@@ -198,7 +202,8 @@ public class ScalingExecutorTest {
metrics = Map.of(jobVertexID, evaluated(1, 110, 101));
assertEquals(
scalingEnabled,
- scalingDecisionExecutor.scaleResource(flinkDep, scalingInfo, conf, metrics));
+ scalingDecisionExecutor.scaleResource(
+ flinkDep, scalingInfo, conf, metrics, kubernetesClient));
var event2 = eventCollector.events.poll();
assertEquals(event.getMetadata().getUid(), event2.getMetadata().getUid());
assertEquals(2, event2.getCount());
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 80ae70bb..2c2c49b9 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
@@ -101,7 +101,7 @@ public class FlinkOperator {
this.listeners = ListenerUtils.discoverListeners(configManager);
this.eventRecorder = EventRecorder.create(client, listeners);
this.ctxFactory =
- new FlinkResourceContextFactory(client, configManager, metricGroup, eventRecorder);
+ new FlinkResourceContextFactory(configManager, metricGroup, eventRecorder);
PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(baseConfig);
FileSystem.initialize(baseConfig, pluginManager);
this.operatorHealthService = OperatorHealthService.fromConfig(configManager);
@@ -163,10 +163,9 @@ public class FlinkOperator {
var autoscalerFactory = AutoscalerLoader.loadJobAutoscalerFactory();
var reconcilerFactory =
new ReconcilerFactory(
- client, configManager, eventRecorder, statusRecorder, autoscalerFactory);
+ configManager, eventRecorder, statusRecorder, autoscalerFactory);
var observerFactory = new FlinkDeploymentObserverFactory(eventRecorder);
- var canaryResourceManager =
- new CanaryResourceManager<FlinkDeployment>(configManager, client);
+ var canaryResourceManager = new CanaryResourceManager<FlinkDeployment>(configManager);
HealthProbe.INSTANCE.registerCanaryResourceManager(canaryResourceManager);
var controller =
@@ -187,10 +186,9 @@ public class FlinkOperator {
var metricManager =
MetricManager.createFlinkSessionJobMetricManager(baseConfig, metricGroup);
var statusRecorder = StatusRecorder.create(client, metricManager, listeners);
- var reconciler = new SessionJobReconciler(client, eventRecorder, statusRecorder);
+ var reconciler = new SessionJobReconciler(eventRecorder, statusRecorder);
var observer = new FlinkSessionJobObserver(eventRecorder);
- var canaryResourceManager =
- new CanaryResourceManager<FlinkSessionJob>(configManager, client);
+ var canaryResourceManager = new CanaryResourceManager<FlinkSessionJob>(configManager);
HealthProbe.INSTANCE.registerCanaryResourceManager(canaryResourceManager);
var controller =
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 2567d6ed..e4fe0a13 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
@@ -99,7 +99,8 @@ public class FlinkDeploymentController
EventRecorder.Type.Normal,
EventRecorder.Reason.Cleanup,
EventRecorder.Component.Operator,
- msg);
+ msg,
+ josdkContext.getClient());
statusRecorder.updateStatusFromCache(flinkApp);
var ctx = ctxFactory.getResourceContext(flinkApp, josdkContext);
try {
@@ -116,7 +117,8 @@ public class FlinkDeploymentController
public UpdateControl<FlinkDeployment> reconcile(FlinkDeployment flinkApp, Context josdkContext)
throws Exception {
- if (canaryResourceManager.handleCanaryResourceReconciliation(flinkApp)) {
+ if (canaryResourceManager.handleCanaryResourceReconciliation(
+ flinkApp, josdkContext.getClient())) {
return UpdateControl.noUpdate();
}
@@ -128,11 +130,11 @@ public class FlinkDeploymentController
try {
observerFactory.getOrCreate(flinkApp).observe(ctx);
if (!validateDeployment(ctx)) {
- statusRecorder.patchAndCacheStatus(flinkApp);
+ statusRecorder.patchAndCacheStatus(flinkApp, ctx.getKubernetesClient());
return ReconciliationUtils.toUpdateControl(
ctx.getOperatorConfig(), flinkApp, previousDeployment, false);
}
- statusRecorder.patchAndCacheStatus(flinkApp);
+ statusRecorder.patchAndCacheStatus(flinkApp, ctx.getKubernetesClient());
reconcilerFactory.getOrCreate(flinkApp).reconcile(ctx);
} catch (RecoveryFailureException rfe) {
handleRecoveryFailed(ctx, rfe);
@@ -144,12 +146,13 @@ public class FlinkDeploymentController
EventRecorder.Type.Warning,
"ClusterDeploymentException",
e.getMessage(),
- EventRecorder.Component.JobManagerDeployment);
+ EventRecorder.Component.JobManagerDeployment,
+ josdkContext.getClient());
throw new ReconciliationException(e);
}
LOG.debug("End of reconciliation");
- statusRecorder.patchAndCacheStatus(flinkApp);
+ statusRecorder.patchAndCacheStatus(flinkApp, ctx.getKubernetesClient());
return ReconciliationUtils.toUpdateControl(
ctx.getOperatorConfig(), flinkApp, previousDeployment, true);
}
@@ -166,7 +169,8 @@ public class FlinkDeploymentController
EventRecorder.Type.Warning,
dfe.getReason(),
dfe.getMessage(),
- EventRecorder.Component.JobManagerDeployment);
+ EventRecorder.Component.JobManagerDeployment,
+ ctx.getKubernetesClient());
}
private void handleRecoveryFailed(
@@ -179,7 +183,8 @@ public class FlinkDeploymentController
EventRecorder.Type.Warning,
rfe.getReason(),
rfe.getMessage(),
- EventRecorder.Component.JobManagerDeployment);
+ EventRecorder.Component.JobManagerDeployment,
+ ctx.getKubernetesClient());
}
@Override
@@ -207,7 +212,8 @@ public class FlinkDeploymentController
EventRecorder.Type.Warning,
EventRecorder.Reason.ValidationError,
EventRecorder.Component.Operator,
- validationError.get());
+ validationError.get(),
+ ctx.getKubernetesClient());
return ReconciliationUtils.applyValidationErrorAndResetSpec(
ctx, validationError.get());
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
index 78b27740..08286a6c 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkResourceContext.java
@@ -27,6 +27,7 @@ import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
import org.apache.flink.kubernetes.operator.metrics.KubernetesResourceMetricGroup;
import org.apache.flink.kubernetes.operator.service.FlinkService;
+import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
@@ -107,4 +108,13 @@ public abstract class FlinkResourceContext<CR extends AbstractFlinkResource<?, ?
configManager.getOperatorConfiguration(
getResource().getMetadata().getNamespace(), getFlinkVersion());
}
+
+ /**
+ * Get KubernetesClient available from JOSDK Context.
+ *
+ * @return KubernetesClient.
+ */
+ public KubernetesClient getKubernetesClient() {
+ return getJosdkContext().getClient();
+ }
}
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 a3ab93cb..34972915 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
@@ -87,7 +87,8 @@ public class FlinkSessionJobController
public UpdateControl<FlinkSessionJob> reconcile(
FlinkSessionJob flinkSessionJob, Context josdkContext) {
- if (canaryResourceManager.handleCanaryResourceReconciliation(flinkSessionJob)) {
+ if (canaryResourceManager.handleCanaryResourceReconciliation(
+ flinkSessionJob, josdkContext.getClient())) {
return UpdateControl.noUpdate();
}
@@ -99,13 +100,13 @@ public class FlinkSessionJobController
observer.observe(ctx);
if (!validateSessionJob(ctx)) {
- statusRecorder.patchAndCacheStatus(flinkSessionJob);
+ statusRecorder.patchAndCacheStatus(flinkSessionJob, ctx.getKubernetesClient());
return ReconciliationUtils.toUpdateControl(
ctx.getOperatorConfig(), flinkSessionJob, previousJob, false);
}
try {
- statusRecorder.patchAndCacheStatus(flinkSessionJob);
+ statusRecorder.patchAndCacheStatus(flinkSessionJob, ctx.getKubernetesClient());
reconciler.reconcile(ctx);
} catch (Exception e) {
eventRecorder.triggerEvent(
@@ -113,10 +114,11 @@ public class FlinkSessionJobController
EventRecorder.Type.Warning,
"SessionJobException",
e.getMessage(),
- EventRecorder.Component.Job);
+ EventRecorder.Component.Job,
+ josdkContext.getClient());
throw new ReconciliationException(e);
}
- statusRecorder.patchAndCacheStatus(flinkSessionJob);
+ statusRecorder.patchAndCacheStatus(flinkSessionJob, ctx.getKubernetesClient());
return ReconciliationUtils.toUpdateControl(
ctx.getOperatorConfig(), flinkSessionJob, previousJob, true);
}
@@ -135,7 +137,8 @@ public class FlinkSessionJobController
EventRecorder.Type.Normal,
EventRecorder.Reason.Cleanup,
EventRecorder.Component.Operator,
- msg);
+ msg,
+ josdkContext.getClient());
statusRecorder.removeCachedStatus(sessionJob);
var ctx = ctxFactory.getResourceContext(sessionJob, josdkContext);
return reconciler.cleanup(ctx);
@@ -168,7 +171,8 @@ public class FlinkSessionJobController
EventRecorder.Type.Warning,
EventRecorder.Reason.ValidationError,
EventRecorder.Component.Operator,
- validationError.get());
+ validationError.get(),
+ ctx.getKubernetesClient());
return ReconciliationUtils.applyValidationErrorAndResetSpec(
ctx, validationError.get());
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManager.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManager.java
index 75a39680..f141239e 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManager.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManager.java
@@ -49,13 +49,12 @@ public class CanaryResourceManager<CR extends AbstractFlinkResource<?, ?>> {
private final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);
private final FlinkConfigManager configManager;
- private final KubernetesClient kubernetesClient;
public boolean allCanariesHealthy() {
return canaryResources.values().stream().allMatch(cr -> cr.isHealthy);
}
- public boolean handleCanaryResourceReconciliation(CR resource) {
+ public boolean handleCanaryResourceReconciliation(CR resource, KubernetesClient client) {
if (!isCanaryResource(resource)) {
return false;
}
@@ -74,7 +73,7 @@ public class CanaryResourceManager<CR extends AbstractFlinkResource<?, ?>> {
}
previousState.onReconcile(resource);
if (firstReconcile) {
- updateSpecAndScheduleHealthCheck(resourceId, previousState);
+ updateSpecAndScheduleHealthCheck(resourceId, previousState, client);
}
return previousState;
});
@@ -93,7 +92,8 @@ public class CanaryResourceManager<CR extends AbstractFlinkResource<?, ?>> {
return true;
}
- private void updateSpecAndScheduleHealthCheck(ResourceID resourceID, CanaryResourceState crs) {
+ private void updateSpecAndScheduleHealthCheck(
+ ResourceID resourceID, CanaryResourceState crs, KubernetesClient client) {
var canaryTimeout =
configManager
.getDefaultConfig()
@@ -106,17 +106,19 @@ public class CanaryResourceManager<CR extends AbstractFlinkResource<?, ?>> {
LOG.info("Scheduling canary check for {} in {}s", resourceID, canaryTimeout.toSeconds());
try {
- kubernetesClient.resource(ReconciliationUtils.clone(crs.resource)).replace();
+ client.resource(ReconciliationUtils.clone(crs.resource)).replace();
} catch (Throwable t) {
LOG.warn("Could not bump canary deployment, it may have been deleted", t);
}
executorService.schedule(
- () -> checkHealth(resourceID), canaryTimeout.toMillis(), TimeUnit.MILLISECONDS);
+ () -> checkHealth(resourceID, client),
+ canaryTimeout.toMillis(),
+ TimeUnit.MILLISECONDS);
}
@VisibleForTesting
- protected void checkHealth(ResourceID resourceID) {
+ protected void checkHealth(ResourceID resourceID, KubernetesClient client) {
CanaryResourceState crs = canaryResources.get(resourceID);
if (crs == null) {
LOG.info("Canary resource {} not found. Stopping health checks", resourceID);
@@ -137,7 +139,7 @@ public class CanaryResourceManager<CR extends AbstractFlinkResource<?, ?>> {
}
// Update spec and reschedule health check
- updateSpecAndScheduleHealthCheck(resourceID, crs);
+ updateSpecAndScheduleHealthCheck(resourceID, crs, client);
}
@VisibleForTesting
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/AbstractFlinkResourceObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/AbstractFlinkResourceObserver.java
index 732dcc60..0d30292b 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/AbstractFlinkResourceObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/AbstractFlinkResourceObserver.java
@@ -48,7 +48,8 @@ public abstract class AbstractFlinkResourceObserver<CR extends AbstractFlinkReso
// Trigger resource specific observe logic
observeInternal(ctx);
- SnapshotUtils.resetSnapshotTriggers(ctx.getResource(), eventRecorder);
+ SnapshotUtils.resetSnapshotTriggers(
+ ctx.getResource(), eventRecorder, ctx.getKubernetesClient());
}
/**
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
index 41e2d03c..e9a80368 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
@@ -178,7 +178,8 @@ public abstract class JobStatusObserver<R extends AbstractFlinkResource<?, ?>> {
EventRecorder.Type.Normal,
EventRecorder.Reason.JobStatusChanged,
EventRecorder.Component.Job,
- message);
+ message,
+ ctx.getKubernetesClient());
}
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java
index 3d3a68a2..51bc9781 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java
@@ -132,7 +132,8 @@ public class SnapshotObserver<
EventRecorder.Reason.SavepointError,
EventRecorder.Component.Operator,
savepointInfo.formatErrorMessage(
- resource.getSpec().getJob().getSavepointTriggerNonce()));
+ resource.getSpec().getJob().getSavepointTriggerNonce()),
+ ctx.getKubernetesClient());
savepointInfo.resetTrigger();
return;
}
@@ -190,7 +191,8 @@ public class SnapshotObserver<
EventRecorder.Reason.CheckpointError,
EventRecorder.Component.Operator,
checkpointInfo.formatErrorMessage(
- resource.getSpec().getJob().getCheckpointTriggerNonce()));
+ resource.getSpec().getJob().getCheckpointTriggerNonce()),
+ ctx.getKubernetesClient());
checkpointInfo.resetTrigger();
return;
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
index 9e732b9e..54c2cd4f 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
@@ -224,7 +224,8 @@ public abstract class AbstractFlinkDeploymentObserver
EventRecorder.Type.Warning,
EventRecorder.Reason.Missing,
EventRecorder.Component.JobManagerDeployment,
- err);
+ err,
+ ctx.getKubernetesClient());
}
@Override
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
index 4472af0c..ba4002e1 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
@@ -111,7 +111,8 @@ public class ApplicationObserver extends AbstractFlinkDeploymentObserver {
EventRecorder.Type.Warning,
EventRecorder.Reason.Missing,
EventRecorder.Component.Job,
- err);
+ err,
+ ctx.getKubernetesClient());
}
}
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserver.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserver.java
index 02291906..d6d7f777 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserver.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserver.java
@@ -193,7 +193,8 @@ public class FlinkSessionJobObserver extends AbstractFlinkResourceObserver<Flink
EventRecorder.Type.Warning,
EventRecorder.Reason.Missing,
EventRecorder.Component.Job,
- MISSING_SESSION_JOB_ERR);
+ MISSING_SESSION_JOB_ERR,
+ ctx.getKubernetesClient());
}
}
}
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 0488c7b9..5bf120bf 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
@@ -444,7 +444,7 @@ public class ReconciliationUtils {
statusRecorder.updateStatusFromCache(ctx.getResource());
ReconciliationUtils.updateForReconciliationError(ctx, e);
- statusRecorder.patchAndCacheStatus(ctx.getResource());
+ statusRecorder.patchAndCacheStatus(ctx.getResource(), ctx.getKubernetesClient());
// Status was updated already, no need to return anything
return ErrorStatusUpdateControl.noStatusUpdate();
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
index 2c5a3a23..0967f60a 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
@@ -76,7 +76,6 @@ public abstract class AbstractFlinkResourceReconciler<
protected final EventRecorder eventRecorder;
protected final StatusRecorder<CR, STATUS> statusRecorder;
- protected final KubernetesClient kubernetesClient;
protected final JobAutoScaler resourceScaler;
public static final String MSG_SUSPENDED = "Suspending existing deployment.";
@@ -88,14 +87,12 @@ public abstract class AbstractFlinkResourceReconciler<
protected Clock clock = Clock.systemDefaultZone();
public AbstractFlinkResourceReconciler(
- KubernetesClient kubernetesClient,
EventRecorder eventRecorder,
StatusRecorder<CR, STATUS> statusRecorder,
JobAutoScalerFactory autoscalerFactory) {
- this.kubernetesClient = kubernetesClient;
this.eventRecorder = eventRecorder;
this.statusRecorder = statusRecorder;
- this.resourceScaler = autoscalerFactory.create(kubernetesClient, eventRecorder);
+ this.resourceScaler = autoscalerFactory.create(eventRecorder);
}
@Override
@@ -122,7 +119,8 @@ public abstract class AbstractFlinkResourceReconciler<
LOG.info("Deploying for the first time");
var deployConfig = ctx.getDeployConfig(spec);
- updateStatusBeforeFirstDeployment(cr, spec, deployConfig, status);
+ updateStatusBeforeFirstDeployment(
+ cr, spec, deployConfig, status, ctx.getKubernetesClient());
deploy(
ctx,
spec,
@@ -160,7 +158,7 @@ public abstract class AbstractFlinkResourceReconciler<
if (checkNewSpecAlreadyDeployed(cr, deployConfig)) {
return;
}
- triggerSpecChangeEvent(cr, specDiff);
+ triggerSpecChangeEvent(cr, specDiff, ctx.getKubernetesClient());
// Try scaling if this is not an upgrade change
boolean scaled = diffType != DiffType.UPGRADE && scale(ctx, deployConfig);
@@ -186,7 +184,8 @@ public abstract class AbstractFlinkResourceReconciler<
EventRecorder.Type.Normal,
EventRecorder.Reason.Rollback,
EventRecorder.Component.JobManagerDeployment,
- MSG_ROLLBACK);
+ MSG_ROLLBACK,
+ ctx.getKubernetesClient());
} else if (!reconcileOtherChanges(ctx)) {
if (resourceScaler.scale(ctx)) {
LOG.info(
@@ -198,14 +197,15 @@ public abstract class AbstractFlinkResourceReconciler<
}
}
- private void triggerSpecChangeEvent(CR cr, DiffResult<SPEC> specDiff) {
+ private void triggerSpecChangeEvent(CR cr, DiffResult<SPEC> specDiff, KubernetesClient client) {
eventRecorder.triggerEventOnce(
cr,
EventRecorder.Type.Normal,
EventRecorder.Reason.SpecChanged,
EventRecorder.Component.JobManagerDeployment,
String.format(MSG_SPEC_CHANGED, specDiff.getType(), specDiff),
- "SpecChange: " + cr.getMetadata().getGeneration());
+ "SpecChange: " + cr.getMetadata().getGeneration(),
+ client);
}
/**
@@ -218,7 +218,7 @@ public abstract class AbstractFlinkResourceReconciler<
* @param status Resource status
*/
private void updateStatusBeforeFirstDeployment(
- CR cr, SPEC spec, Configuration deployConfig, STATUS status) {
+ CR cr, SPEC spec, Configuration deployConfig, STATUS status, KubernetesClient client) {
if (spec.getJob() != null) {
var initialUpgradeMode = UpgradeMode.STATELESS;
var initialSp = spec.getJob().getInitialSavepointPath();
@@ -235,7 +235,7 @@ public abstract class AbstractFlinkResourceReconciler<
ReconciliationUtils.updateStatusBeforeDeploymentAttempt(cr, deployConfig, clock);
// Before we try to submit the job we record the current spec in the status so we can
// handle subsequent deployment and status update errors
- statusRecorder.patchAndCacheStatus(cr);
+ statusRecorder.patchAndCacheStatus(cr, client);
}
/**
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
index 1860da5e..6db4905c 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
@@ -35,7 +35,6 @@ import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.SnapshotUtils;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
-import io.fabric8.kubernetes.client.KubernetesClient;
import lombok.Value;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,11 +59,10 @@ public abstract class AbstractJobReconciler<
private static final Logger LOG = LoggerFactory.getLogger(AbstractJobReconciler.class);
public AbstractJobReconciler(
- KubernetesClient kubernetesClient,
EventRecorder eventRecorder,
StatusRecorder<CR, STATUS> statusRecorder,
JobAutoScalerFactory autoscalerFactory) {
- super(kubernetesClient, eventRecorder, statusRecorder, autoscalerFactory);
+ super(eventRecorder, statusRecorder, autoscalerFactory);
}
@Override
@@ -112,7 +110,8 @@ public abstract class AbstractJobReconciler<
EventRecorder.Type.Normal,
EventRecorder.Reason.Suspended,
EventRecorder.Component.JobManagerDeployment,
- MSG_SUSPENDED);
+ MSG_SUSPENDED,
+ ctx.getKubernetesClient());
UpgradeMode upgradeMode = availableUpgradeMode.getUpgradeMode().get();
@@ -137,7 +136,7 @@ public abstract class AbstractJobReconciler<
}
// We record the target spec into an upgrading state before deploying
ReconciliationUtils.updateStatusBeforeDeploymentAttempt(resource, deployConfig, clock);
- statusRecorder.patchAndCacheStatus(resource);
+ statusRecorder.patchAndCacheStatus(resource, ctx.getKubernetesClient());
restoreJob(
ctx,
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
index bf80e287..5ba2154a 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
@@ -65,11 +65,10 @@ public class ApplicationReconciler
static final String MSG_RESTART_UNHEALTHY = "Restarting unhealthy job";
public ApplicationReconciler(
- KubernetesClient kubernetesClient,
EventRecorder eventRecorder,
StatusRecorder<FlinkDeployment, FlinkDeploymentStatus> statusRecorder,
JobAutoScalerFactory autoscalerFactory) {
- super(kubernetesClient, eventRecorder, statusRecorder, autoscalerFactory);
+ super(eventRecorder, statusRecorder, autoscalerFactory);
}
@Override
@@ -168,27 +167,31 @@ public class ApplicationReconciler
flinkService.deleteClusterDeployment(
relatedResource.getMetadata(), status, deployConfig, true);
flinkService.waitForClusterShutdown(deployConfig);
- statusRecorder.patchAndCacheStatus(relatedResource);
+ statusRecorder.patchAndCacheStatus(relatedResource, ctx.getKubernetesClient());
}
- setJobIdIfNecessary(spec, relatedResource, deployConfig);
+ setJobIdIfNecessary(spec, relatedResource, deployConfig, ctx.getKubernetesClient());
eventRecorder.triggerEvent(
relatedResource,
EventRecorder.Type.Normal,
EventRecorder.Reason.Submit,
EventRecorder.Component.JobManagerDeployment,
- MSG_SUBMIT);
+ MSG_SUBMIT,
+ ctx.getKubernetesClient());
flinkService.submitApplicationCluster(spec.getJob(), deployConfig, requireHaMetadata);
status.getJobStatus().setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
IngressUtils.updateIngressRules(
- relatedResource.getMetadata(), spec, deployConfig, kubernetesClient);
+ relatedResource.getMetadata(), spec, deployConfig, ctx.getKubernetesClient());
}
private void setJobIdIfNecessary(
- FlinkDeploymentSpec spec, FlinkDeployment resource, Configuration deployConfig) {
+ FlinkDeploymentSpec spec,
+ FlinkDeployment resource,
+ Configuration deployConfig,
+ KubernetesClient client) {
// The jobId assigned by Flink would be constant,
// overwrite to avoid checkpoint path conflicts.
// https://issues.apache.org/jira/browse/FLINK-19358
@@ -207,7 +210,7 @@ public class ApplicationReconciler
// record before first deployment to ensure we use it on any retry
status.getJobStatus().setJobId(jobId);
LOG.info("Assigning JobId override to {}", jobId);
- statusRecorder.patchAndCacheStatus(resource);
+ statusRecorder.patchAndCacheStatus(resource, client);
}
String jobId = status.getJobStatus().getJobId();
@@ -268,7 +271,8 @@ public class ApplicationReconciler
EventRecorder.Type.Warning,
EventRecorder.Reason.RecoverDeployment,
EventRecorder.Component.Job,
- MSG_RECOVERY);
+ MSG_RECOVERY,
+ ctx.getKubernetesClient());
}
if (shouldRestartJobBecauseUnhealthy) {
@@ -277,7 +281,8 @@ public class ApplicationReconciler
EventRecorder.Type.Warning,
EventRecorder.Reason.RestartUnhealthyJob,
EventRecorder.Component.Job,
- MSG_RESTART_UNHEALTHY);
+ MSG_RESTART_UNHEALTHY,
+ ctx.getKubernetesClient());
cleanupAfterFailedJob(ctx);
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/JobAutoScalerFactory.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/JobAutoScalerFactory.java
index 927657ad..c390a9d2 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/JobAutoScalerFactory.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/JobAutoScalerFactory.java
@@ -19,10 +19,8 @@ package org.apache.flink.kubernetes.operator.reconciler.deployment;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
-import io.fabric8.kubernetes.client.KubernetesClient;
-
/** Factory to construct a new autoscaler instance. */
public interface JobAutoScalerFactory {
- JobAutoScaler create(KubernetesClient kubernetesClient, EventRecorder eventRecorder);
+ JobAutoScaler create(EventRecorder eventRecorder);
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/NoopJobAutoscalerFactory.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/NoopJobAutoscalerFactory.java
index 5e467257..f819123d 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/NoopJobAutoscalerFactory.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/NoopJobAutoscalerFactory.java
@@ -20,15 +20,13 @@ package org.apache.flink.kubernetes.operator.reconciler.deployment;
import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
-import io.fabric8.kubernetes.client.KubernetesClient;
-
import java.util.Map;
/** An autoscaler implementation which does nothing. */
public class NoopJobAutoscalerFactory implements JobAutoScalerFactory, JobAutoScaler {
@Override
- public JobAutoScaler create(KubernetesClient kubernetesClient, EventRecorder eventRecorder) {
+ public JobAutoScaler create(EventRecorder eventRecorder) {
return this;
}
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 d1b3863e..85aede5a 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
@@ -27,15 +27,12 @@ import org.apache.flink.kubernetes.operator.reconciler.Reconciler;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
-import io.fabric8.kubernetes.client.KubernetesClient;
-
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/** The factory to create reconciler based on app mode. */
public class ReconcilerFactory {
- private final KubernetesClient kubernetesClient;
private final FlinkConfigManager configManager;
private final EventRecorder eventRecorder;
private final StatusRecorder<FlinkDeployment, FlinkDeploymentStatus> deploymentStatusRecorder;
@@ -44,12 +41,10 @@ public class ReconcilerFactory {
reconcilerMap;
public ReconcilerFactory(
- KubernetesClient kubernetesClient,
FlinkConfigManager configManager,
EventRecorder eventRecorder,
StatusRecorder<FlinkDeployment, FlinkDeploymentStatus> deploymentStatusRecorder,
JobAutoScalerFactory autoscalerFactory) {
- this.kubernetesClient = kubernetesClient;
this.configManager = configManager;
this.eventRecorder = eventRecorder;
this.deploymentStatusRecorder = deploymentStatusRecorder;
@@ -65,14 +60,10 @@ public class ReconcilerFactory {
modes -> {
switch (modes.f0) {
case SESSION:
- return new SessionReconciler(
- kubernetesClient, eventRecorder, deploymentStatusRecorder);
+ return new SessionReconciler(eventRecorder, deploymentStatusRecorder);
case APPLICATION:
return new ApplicationReconciler(
- kubernetesClient,
- eventRecorder,
- deploymentStatusRecorder,
- autoscalerFactory);
+ eventRecorder, deploymentStatusRecorder, autoscalerFactory);
default:
throw new UnsupportedOperationException(
String.format("Unsupported running mode: %s", modes.f0));
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 0154e639..1f6d2cbd 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
@@ -29,7 +29,6 @@ import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.IngressUtils;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
-import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,10 +48,9 @@ public class SessionReconciler
private static final Logger LOG = LoggerFactory.getLogger(SessionReconciler.class);
public SessionReconciler(
- KubernetesClient kubernetesClient,
EventRecorder eventRecorder,
StatusRecorder<FlinkDeployment, FlinkDeploymentStatus> statusRecorder) {
- super(kubernetesClient, eventRecorder, statusRecorder, new NoopJobAutoscalerFactory());
+ super(eventRecorder, statusRecorder, new NoopJobAutoscalerFactory());
}
@Override
@@ -69,7 +67,7 @@ public class SessionReconciler
// We record the target spec into an upgrading state before deploying
ReconciliationUtils.updateStatusBeforeDeploymentAttempt(deployment, deployConfig, clock);
- statusRecorder.patchAndCacheStatus(deployment);
+ statusRecorder.patchAndCacheStatus(deployment, ctx.getKubernetesClient());
deploy(ctx, deployment.getSpec(), deployConfig, Optional.empty(), false);
ReconciliationUtils.updateStatusForDeployedSpec(deployment, deployConfig, clock);
@@ -97,7 +95,8 @@ public class SessionReconciler
setOwnerReference(cr, deployConfig);
ctx.getFlinkService().submitSessionCluster(deployConfig);
cr.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
- IngressUtils.updateIngressRules(cr.getMetadata(), spec, deployConfig, kubernetesClient);
+ IngressUtils.updateIngressRules(
+ cr.getMetadata(), spec, deployConfig, ctx.getKubernetesClient());
}
@Override
@@ -134,7 +133,8 @@ public class SessionReconciler
EventRecorder.Type.Warning,
EventRecorder.Reason.CleanupFailed,
EventRecorder.Component.Operator,
- error)) {
+ error,
+ ctx.getKubernetesClient())) {
LOG.warn(error);
}
return DeleteControl.noFinalizerRemoval()
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
index 6ac7c373..9521d841 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconciler.java
@@ -33,7 +33,6 @@ import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.apache.flink.runtime.messages.FlinkJobNotFoundException;
import org.apache.flink.runtime.messages.FlinkJobTerminatedWithoutCancellationException;
-import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,10 +47,9 @@ public class SessionJobReconciler
private static final Logger LOG = LoggerFactory.getLogger(SessionJobReconciler.class);
public SessionJobReconciler(
- KubernetesClient kubernetesClient,
EventRecorder eventRecorder,
StatusRecorder<FlinkSessionJob, FlinkSessionJobStatus> statusRecorder) {
- super(kubernetesClient, eventRecorder, statusRecorder, new NoopJobAutoscalerFactory());
+ super(eventRecorder, statusRecorder, new NoopJobAutoscalerFactory());
}
@Override
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java
index 3698ed0b..ae331348 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java
@@ -33,7 +33,6 @@ import org.apache.flink.kubernetes.operator.metrics.OperatorMetricUtils;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
-import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import org.slf4j.Logger;
@@ -49,7 +48,6 @@ public class FlinkResourceContextFactory {
private static final Logger LOG = LoggerFactory.getLogger(FlinkResourceContextFactory.class);
- private final KubernetesClient kubernetesClient;
private final FlinkConfigManager configManager;
private final ArtifactManager artifactManager;
private final ExecutorService clientExecutorService;
@@ -60,11 +58,9 @@ public class FlinkResourceContextFactory {
resourceMetricGroups = new ConcurrentHashMap<>();
public FlinkResourceContextFactory(
- KubernetesClient kubernetesClient,
FlinkConfigManager configManager,
KubernetesOperatorMetricGroup operatorMetricGroup,
EventRecorder eventRecorder) {
- this.kubernetesClient = kubernetesClient;
this.configManager = configManager;
this.operatorMetricGroup = operatorMetricGroup;
this.eventRecorder = eventRecorder;
@@ -109,14 +105,14 @@ public class FlinkResourceContextFactory {
switch (deploymentMode) {
case NATIVE:
return new NativeFlinkService(
- kubernetesClient,
+ ctx.getKubernetesClient(),
artifactManager,
clientExecutorService,
ctx.getOperatorConfig(),
eventRecorder);
case STANDALONE:
return new StandaloneFlinkService(
- kubernetesClient,
+ ctx.getKubernetesClient(),
artifactManager,
clientExecutorService,
ctx.getOperatorConfig());
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java
index 7c0ed976..0f55e948 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/NativeFlinkService.java
@@ -230,7 +230,8 @@ public class NativeFlinkService extends AbstractFlinkService {
EventRecorder.Type.Normal,
EventRecorder.Reason.Scaling,
EventRecorder.Component.Job,
- "In-place scaling triggered");
+ "In-place scaling triggered",
+ ctx.getKubernetesClient());
}
return result;
} catch (Throwable t) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
index 98644d23..bc5eab3c 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventRecorder.java
@@ -34,12 +34,9 @@ import java.util.function.BiConsumer;
/** Helper class for creating Kubernetes events for Flink resources. */
public class EventRecorder {
- private final KubernetesClient client;
private final BiConsumer<AbstractFlinkResource<?, ?>, Event> eventListener;
- public EventRecorder(
- KubernetesClient client, BiConsumer<AbstractFlinkResource<?, ?>, Event> eventListener) {
- this.client = client;
+ public EventRecorder(BiConsumer<AbstractFlinkResource<?, ?>, Event> eventListener) {
this.eventListener = eventListener;
}
@@ -48,8 +45,9 @@ public class EventRecorder {
Type type,
Reason reason,
Component component,
- String message) {
- return triggerEvent(resource, type, reason, component, message, null);
+ String message,
+ KubernetesClient client) {
+ return triggerEvent(resource, type, reason, component, message, null, client);
}
public boolean triggerEventOnce(
@@ -58,8 +56,10 @@ public class EventRecorder {
Reason reason,
Component component,
String message,
- String messageKey) {
- return triggerEventOnce(resource, type, reason.toString(), message, component, messageKey);
+ String messageKey,
+ KubernetesClient client) {
+ return triggerEventOnce(
+ resource, type, reason.toString(), message, component, messageKey, client);
}
public boolean triggerEvent(
@@ -68,8 +68,10 @@ public class EventRecorder {
Reason reason,
Component component,
String message,
- @Nullable String messageKey) {
- return triggerEvent(resource, type, reason.toString(), message, component, messageKey);
+ @Nullable String messageKey,
+ KubernetesClient client) {
+ return triggerEvent(
+ resource, type, reason.toString(), message, component, messageKey, client);
}
public boolean triggerEvent(
@@ -78,7 +80,8 @@ public class EventRecorder {
String reason,
String message,
Component component,
- String messageKey) {
+ String messageKey,
+ KubernetesClient client) {
return EventUtils.createOrUpdateEvent(
client,
resource,
@@ -96,7 +99,8 @@ public class EventRecorder {
String reason,
String message,
Component component,
- String messageKey) {
+ String messageKey,
+ KubernetesClient client) {
return EventUtils.createIfNotExists(
client,
resource,
@@ -113,8 +117,9 @@ public class EventRecorder {
Type type,
String reason,
String message,
- Component component) {
- return triggerEvent(resource, type, reason, message, component, null);
+ Component component,
+ KubernetesClient client) {
+ return triggerEvent(resource, type, reason, message, component, null, client);
}
public static EventRecorder create(
@@ -150,7 +155,7 @@ public class EventRecorder {
AuditUtils.logContext(ctx);
};
- return new EventRecorder(client, biConsumer);
+ return new EventRecorder(biConsumer);
}
/** The type of the events. */
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SnapshotUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SnapshotUtils.java
index 4cc5ee2f..fdd537d6 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SnapshotUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SnapshotUtils.java
@@ -31,6 +31,7 @@ import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
import org.apache.flink.kubernetes.operator.service.FlinkService;
+import io.fabric8.kubernetes.client.KubernetesClient;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.core.util.CronExpression;
import org.slf4j.Logger;
@@ -363,7 +364,9 @@ public class SnapshotUtils {
}
public static void resetSnapshotTriggers(
- AbstractFlinkResource<?, ?> resource, EventRecorder eventRecorder) {
+ AbstractFlinkResource<?, ?> resource,
+ EventRecorder eventRecorder,
+ KubernetesClient client) {
var status = resource.getStatus();
var jobStatus = status.getJobStatus();
@@ -380,7 +383,8 @@ public class SnapshotUtils {
EventRecorder.Reason.SavepointError,
EventRecorder.Component.Operator,
savepointInfo.formatErrorMessage(
- resource.getSpec().getJob().getSavepointTriggerNonce()));
+ resource.getSpec().getJob().getSavepointTriggerNonce()),
+ client);
}
if (SnapshotUtils.checkpointInProgress(jobStatus)) {
var checkpointInfo = jobStatus.getCheckpointInfo();
@@ -394,7 +398,8 @@ public class SnapshotUtils {
EventRecorder.Reason.CheckpointError,
EventRecorder.Component.Operator,
checkpointInfo.formatErrorMessage(
- resource.getSpec().getJob().getCheckpointTriggerNonce()));
+ resource.getSpec().getJob().getCheckpointTriggerNonce()),
+ client);
}
}
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
index e009a7ce..0197dadd 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java
@@ -55,15 +55,11 @@ public class StatusRecorder<
protected final ConcurrentHashMap<ResourceID, ObjectNode> statusCache =
new ConcurrentHashMap<>();
- private final KubernetesClient client;
private final MetricManager<CR> metricManager;
private final BiConsumer<CR, STATUS> statusUpdateListener;
public StatusRecorder(
- KubernetesClient client,
- MetricManager<CR> metricManager,
- BiConsumer<CR, STATUS> statusUpdateListener) {
- this.client = client;
+ MetricManager<CR> metricManager, BiConsumer<CR, STATUS> statusUpdateListener) {
this.statusUpdateListener = statusUpdateListener;
this.metricManager = metricManager;
}
@@ -77,7 +73,7 @@ public class StatusRecorder<
* @param resource Resource for which status update should be performed
*/
@SneakyThrows
- public void patchAndCacheStatus(CR resource) {
+ public void patchAndCacheStatus(CR resource, KubernetesClient client) {
ObjectNode newStatusNode =
objectMapper.convertValue(resource.getStatus(), ObjectNode.class);
var resourceId = ResourceID.fromResource(resource);
@@ -98,7 +94,7 @@ public class StatusRecorder<
for (int i = 0; i < 3; i++) {
// We retry the status update 3 times to avoid some intermittent connectivity errors
try {
- replaceStatus(resource, prevStatus);
+ replaceStatus(resource, prevStatus, client);
err = null;
} catch (KubernetesClientException e) {
LOG.error("Error while patching status, retrying {}/3...", (i + 1), e);
@@ -116,7 +112,8 @@ public class StatusRecorder<
metricManager.onUpdate(resource);
}
- private void replaceStatus(CR resource, STATUS prevStatus) throws JsonProcessingException {
+ private void replaceStatus(CR resource, STATUS prevStatus, KubernetesClient client)
+ throws JsonProcessingException {
int retries = 0;
while (true) {
try {
@@ -251,6 +248,6 @@ public class StatusRecorder<
AuditUtils.logContext(ctx);
};
- return new StatusRecorder<>(kubernetesClient, metricManager, consumer);
+ return new StatusRecorder<>(metricManager, consumer);
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/OperatorTestBase.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/OperatorTestBase.java
index a8225424..cb9aba35 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/OperatorTestBase.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/OperatorTestBase.java
@@ -48,7 +48,7 @@ public abstract class OperatorTestBase {
getKubernetesClient().resource(TestUtils.buildSessionJob()).createOrReplace();
flinkService = new TestingFlinkService(getKubernetesClient());
context = flinkService.getContext();
- eventRecorder = new EventRecorder(getKubernetesClient(), eventCollector);
+ eventRecorder = new EventRecorder(eventCollector);
operatorMetricGroup = TestUtils.createTestMetricGroup(configManager.getDefaultConfig());
setup();
}
@@ -66,11 +66,7 @@ public abstract class OperatorTestBase {
CR cr, Context josdkContext) {
var ctxFactory =
new TestingFlinkResourceContextFactory(
- getKubernetesClient(),
- configManager,
- operatorMetricGroup,
- flinkService,
- eventRecorder);
+ configManager, operatorMetricGroup, flinkService, eventRecorder);
return ctxFactory.getResourceContext(cr, josdkContext);
}
}
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 e228da2f..d9f58781 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
@@ -147,33 +147,55 @@ public class TestUtils extends BaseTestUtils {
}
public static <T extends HasMetadata> Context<T> createContextWithDeployment(
- @Nullable Deployment deployment) {
+ @Nullable Deployment deployment, KubernetesClient client) {
return new TestingContext<>() {
@Override
public Optional<T> getSecondaryResource(Class expectedType, String eventSourceName) {
return (Optional<T>) Optional.ofNullable(deployment);
}
+
+ @Override
+ public KubernetesClient getClient() {
+ return client;
+ }
};
}
public static <T extends HasMetadata> Context<T> createEmptyContext() {
- return createContextWithDeployment(null);
+ return createContextWithDeployment(null, null);
+ }
+
+ public static <T extends HasMetadata> Context<T> createEmptyContextWithClient(
+ KubernetesClient client) {
+ return createContextWithDeployment(null, client);
}
- public static <T extends HasMetadata> Context<T> createContextWithReadyJobManagerDeployment() {
- return createContextWithDeployment(createDeployment(true));
+ public static <T extends HasMetadata> Context<T> createContextWithReadyJobManagerDeployment(
+ KubernetesClient client) {
+ return createContextWithDeployment(createDeployment(true), client);
}
- public static <T extends HasMetadata> Context<T> createContextWithInProgressDeployment() {
- return createContextWithDeployment(createDeployment(false));
+ public static <T extends HasMetadata> Context<T> createContextWithInProgressDeployment(
+ KubernetesClient client) {
+ return createContextWithDeployment(createDeployment(false), client);
}
public static <T extends HasMetadata> Context<T> createContextWithReadyFlinkDeployment() {
- return createContextWithReadyFlinkDeployment(new HashMap<>());
+ return createContextWithReadyFlinkDeployment(new HashMap<>(), null);
+ }
+
+ public static <T extends HasMetadata> Context<T> createContextWithReadyFlinkDeployment(
+ KubernetesClient client) {
+ return createContextWithReadyFlinkDeployment(new HashMap<>(), client);
}
public static <T extends HasMetadata> Context<T> createContextWithReadyFlinkDeployment(
Map<String, String> flinkDepConfig) {
+ return createContextWithReadyFlinkDeployment(flinkDepConfig, null);
+ }
+
+ public static <T extends HasMetadata> Context<T> createContextWithReadyFlinkDeployment(
+ Map<String, String> flinkDepConfig, KubernetesClient client) {
return new TestingContext<>() {
@Override
public Optional<T> getSecondaryResource(Class expectedType, String eventSourceName) {
@@ -185,6 +207,11 @@ public class TestUtils extends BaseTestUtils {
.serializeAndSetLastReconciledSpec(session.getSpec(), session);
return (Optional<T>) Optional.of(session);
}
+
+ @Override
+ public KubernetesClient getClient() {
+ return client;
+ }
};
}
@@ -202,7 +229,8 @@ public class TestUtils extends BaseTestUtils {
public static final String DEPLOYMENT_ERROR = "test deployment error message";
- public static <T extends HasMetadata> Context<T> createContextWithFailedJobManagerDeployment() {
+ public static <T extends HasMetadata> Context<T> createContextWithFailedJobManagerDeployment(
+ KubernetesClient client) {
return new TestingContext<>() {
@Override
public Optional getSecondaryResource(Class expectedType, String eventSourceName) {
@@ -226,6 +254,11 @@ public class TestUtils extends BaseTestUtils {
deployment.setStatus(status);
return Optional.of(deployment);
}
+
+ @Override
+ public KubernetesClient getClient() {
+ return client;
+ }
};
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkResourceContextFactory.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkResourceContextFactory.java
index ab103b9a..95bf1081 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkResourceContextFactory.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkResourceContextFactory.java
@@ -26,7 +26,6 @@ import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
-import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import java.util.Map;
@@ -36,12 +35,11 @@ public class TestingFlinkResourceContextFactory extends FlinkResourceContextFact
private final FlinkService flinkService;
public TestingFlinkResourceContextFactory(
- KubernetesClient kubernetesClient,
FlinkConfigManager configManager,
KubernetesOperatorMetricGroup operatorMetricGroup,
FlinkService flinkService,
EventRecorder eventRecorder) {
- super(kubernetesClient, configManager, operatorMetricGroup, eventRecorder);
+ super(configManager, operatorMetricGroup, eventRecorder);
this.flinkService = flinkService;
}
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 f5e1e4ec..4dc8e75d 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
@@ -171,6 +171,11 @@ public class TestingFlinkService extends AbstractFlinkService {
}
return (Optional<T>) Optional.of(TestUtils.createDeployment(jobManagerReady));
}
+
+ @Override
+ public KubernetesClient getClient() {
+ return kubernetesClient;
+ }
};
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusRecorder.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusRecorder.java
index 7cd010e3..3fa5fdff 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusRecorder.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusRecorder.java
@@ -24,6 +24,7 @@ import org.apache.flink.kubernetes.operator.metrics.MetricManager;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
/** Testing statusRecorder. */
@@ -32,11 +33,11 @@ public class TestingStatusRecorder<
extends StatusRecorder<CR, STATUS> {
public TestingStatusRecorder() {
- super(null, new MetricManager<>(), (r, s) -> {});
+ super(new MetricManager<>(), (r, s) -> {});
}
@Override
- public void patchAndCacheStatus(CR resource) {
+ public void patchAndCacheStatus(CR resource, KubernetesClient client) {
statusCache.put(
ResourceID.fromResource(resource),
objectMapper.convertValue(resource.getStatus(), ObjectNode.class));
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 1030e461..61592364 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
@@ -54,8 +54,7 @@ public class DeploymentRecoveryTest {
public void setup() {
flinkService = new TestingFlinkService(kubernetesClient);
context = flinkService.getContext();
- testController =
- new TestingFlinkDeploymentController(configManager, kubernetesClient, flinkService);
+ testController = new TestingFlinkDeploymentController(configManager, flinkService);
kubernetesClient.resource(TestUtils.buildApplicationCluster()).createOrReplace();
}
@@ -80,9 +79,11 @@ public class DeploymentRecoveryTest {
// Make sure we do not try to recover JM deployment errors (only missing)
testController.reconcile(
- appCluster, TestUtils.createContextWithFailedJobManagerDeployment());
+ appCluster,
+ TestUtils.createContextWithFailedJobManagerDeployment(kubernetesClient));
testController.reconcile(
- appCluster, TestUtils.createContextWithFailedJobManagerDeployment());
+ appCluster,
+ TestUtils.createContextWithFailedJobManagerDeployment(kubernetesClient));
assertEquals(
JobManagerDeploymentStatus.ERROR,
appCluster.getStatus().getJobManagerDeploymentStatus());
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FailedDeploymentRestartTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FailedDeploymentRestartTest.java
index de79a0db..cdb77fd1 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FailedDeploymentRestartTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FailedDeploymentRestartTest.java
@@ -54,8 +54,7 @@ public class FailedDeploymentRestartTest {
configManager = new FlinkConfigManager(configuration);
flinkService = new TestingFlinkService(kubernetesClient);
context = flinkService.getContext();
- testController =
- new TestingFlinkDeploymentController(configManager, kubernetesClient, flinkService);
+ testController = new TestingFlinkDeploymentController(configManager, flinkService);
kubernetesClient.resource(TestUtils.buildApplicationCluster()).createOrReplace();
}
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 cd1c974d..e9a13160 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
@@ -100,8 +100,7 @@ public class FlinkDeploymentControllerTest {
public void setup() {
flinkService = new TestingFlinkService(kubernetesClient);
context = flinkService.getContext();
- testController =
- new TestingFlinkDeploymentController(configManager, kubernetesClient, flinkService);
+ testController = new TestingFlinkDeploymentController(configManager, flinkService);
kubernetesClient.resource(TestUtils.buildApplicationCluster()).createOrReplace();
}
@@ -250,7 +249,8 @@ public class FlinkDeploymentControllerTest {
testController.reconcile(appCluster, context);
updateControl =
testController.reconcile(
- appCluster, TestUtils.createContextWithFailedJobManagerDeployment());
+ appCluster,
+ TestUtils.createContextWithFailedJobManagerDeployment(kubernetesClient));
submittedEventValidatingResponseProvider.assertValidated();
assertFalse(updateControl.isUpdateStatus());
assertEquals(
@@ -270,7 +270,8 @@ public class FlinkDeploymentControllerTest {
// next cycle should not create another event
updateControl =
testController.reconcile(
- appCluster, TestUtils.createContextWithFailedJobManagerDeployment());
+ appCluster,
+ TestUtils.createContextWithFailedJobManagerDeployment(kubernetesClient));
assertEquals(
JobManagerDeploymentStatus.ERROR,
appCluster.getStatus().getJobManagerDeploymentStatus());
@@ -334,7 +335,8 @@ public class FlinkDeploymentControllerTest {
testController.reconcile(appCluster, context);
updateControl =
testController.reconcile(
- appCluster, TestUtils.createContextWithInProgressDeployment());
+ appCluster,
+ TestUtils.createContextWithInProgressDeployment(kubernetesClient));
submittedEventValidatingResponseProvider.assertValidated();
assertFalse(updateControl.isUpdateStatus());
assertEquals(
@@ -355,7 +357,8 @@ public class FlinkDeploymentControllerTest {
// next cycle should not create another event
updateControl =
testController.reconcile(
- appCluster, TestUtils.createContextWithFailedJobManagerDeployment());
+ appCluster,
+ TestUtils.createContextWithFailedJobManagerDeployment(kubernetesClient));
assertEquals(
JobManagerDeploymentStatus.ERROR,
appCluster.getStatus().getJobManagerDeploymentStatus());
@@ -582,7 +585,8 @@ public class FlinkDeploymentControllerTest {
.getJob()
.getState());
appCluster.getSpec().setLogConfiguration(Map.of("invalid", "conf"));
- testController.reconcile(appCluster, TestUtils.createEmptyContext());
+ testController.reconcile(
+ appCluster, TestUtils.createEmptyContextWithClient(kubernetesClient));
assertEquals(2, testController.events().size());
testController.events().remove();
assertEquals(
@@ -910,7 +914,8 @@ public class FlinkDeploymentControllerTest {
FlinkDeployment appCluster = TestUtils.buildApplicationCluster();
testController.reconcile(appCluster, context);
- testController.reconcile(appCluster, TestUtils.createContextWithInProgressDeployment());
+ testController.reconcile(
+ appCluster, TestUtils.createContextWithInProgressDeployment(kubernetesClient));
assertNull(appCluster.getStatus().getReconciliationStatus().getLastStableSpec());
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobControllerTest.java
index 552f15e8..c7eed4c9 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobControllerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkSessionJobControllerTest.java
@@ -58,7 +58,7 @@ import static org.junit.jupiter.api.Assertions.fail;
class FlinkSessionJobControllerTest {
private KubernetesClient kubernetesClient;
private final FlinkConfigManager configManager = new FlinkConfigManager(new Configuration());
- private final Context context = TestUtils.createContextWithReadyFlinkDeployment();
+ private Context context;
private TestingFlinkService flinkService = new TestingFlinkService();
private TestingFlinkSessionJobController testController;
@@ -68,11 +68,11 @@ class FlinkSessionJobControllerTest {
@BeforeEach
public void before() {
flinkService = new TestingFlinkService();
- testController =
- new TestingFlinkSessionJobController(configManager, kubernetesClient, flinkService);
+ testController = new TestingFlinkSessionJobController(configManager, flinkService);
sessionJob = TestUtils.buildSessionJob();
suspendedSessionJob = TestUtils.buildSessionJob(JobState.SUSPENDED);
kubernetesClient.resource(sessionJob).createOrReplace();
+ context = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
}
@Test
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
index c203339b..667a13c8 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
@@ -70,9 +70,7 @@ public class RollbackTest {
context = flinkService.getContext();
testController =
new TestingFlinkDeploymentController(
- new FlinkConfigManager(new Configuration()),
- kubernetesClient,
- flinkService);
+ new FlinkConfigManager(new Configuration()), flinkService);
kubernetesClient.resource(TestUtils.buildApplicationCluster()).createOrReplace();
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
index b38f6701..e1f3d0ef 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
@@ -38,7 +38,6 @@ import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
import io.fabric8.kubernetes.api.model.Event;
-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.DeleteControl;
@@ -80,29 +79,24 @@ public class TestingFlinkDeploymentController
private Map<ResourceID, Tuple2<FlinkDeploymentSpec, Long>> currentGenerations = new HashMap<>();
public TestingFlinkDeploymentController(
- FlinkConfigManager configManager,
- KubernetesClient kubernetesClient,
- TestingFlinkService flinkService) {
+ FlinkConfigManager configManager, TestingFlinkService flinkService) {
contextFactory =
new TestingFlinkResourceContextFactory(
- kubernetesClient,
configManager,
TestUtils.createTestMetricGroup(new Configuration()),
flinkService,
eventRecorder);
- eventRecorder = new EventRecorder(kubernetesClient, eventCollector);
- statusRecorder =
- new StatusRecorder<>(kubernetesClient, new MetricManager<>(), statusUpdateCounter);
+ eventRecorder = new EventRecorder(eventCollector);
+ statusRecorder = new StatusRecorder<>(new MetricManager<>(), statusUpdateCounter);
reconcilerFactory =
new ReconcilerFactory(
- kubernetesClient,
configManager,
eventRecorder,
statusRecorder,
new NoopJobAutoscalerFactory());
- canaryResourceManager = new CanaryResourceManager<>(configManager, kubernetesClient);
+ canaryResourceManager = new CanaryResourceManager<>(configManager);
flinkDeploymentController =
new FlinkDeploymentController(
ValidatorUtils.discoverValidators(configManager),
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkSessionJobController.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkSessionJobController.java
index bdc45034..8926492c 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkSessionJobController.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkSessionJobController.java
@@ -37,7 +37,6 @@ import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
import io.fabric8.kubernetes.api.model.Event;
-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.DeleteControl;
@@ -73,29 +72,25 @@ public class TestingFlinkSessionJobController
private Map<ResourceID, Tuple2<FlinkSessionJobSpec, Long>> currentGenerations = new HashMap<>();
public TestingFlinkSessionJobController(
- FlinkConfigManager configManager,
- KubernetesClient kubernetesClient,
- TestingFlinkService flinkService) {
+ FlinkConfigManager configManager, TestingFlinkService flinkService) {
var ctxFactory =
new TestingFlinkResourceContextFactory(
- kubernetesClient,
configManager,
TestUtils.createTestMetricGroup(new Configuration()),
flinkService,
eventRecorder);
- eventRecorder = new EventRecorder(kubernetesClient, eventCollector);
+ eventRecorder = new EventRecorder(eventCollector);
- statusRecorder =
- new StatusRecorder<>(kubernetesClient, new MetricManager<>(), statusUpdateCounter);
+ statusRecorder = new StatusRecorder<>(new MetricManager<>(), statusUpdateCounter);
- canaryResourceManager = new CanaryResourceManager<>(configManager, kubernetesClient);
+ canaryResourceManager = new CanaryResourceManager<>(configManager);
flinkSessionJobController =
new FlinkSessionJobController(
ValidatorUtils.discoverValidators(configManager),
ctxFactory,
- new SessionJobReconciler(kubernetesClient, eventRecorder, statusRecorder),
+ new SessionJobReconciler(eventRecorder, statusRecorder),
new FlinkSessionJobObserver(eventRecorder),
statusRecorder,
eventRecorder,
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/UnhealthyDeploymentRestartTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/UnhealthyDeploymentRestartTest.java
index e643c0d2..1d377e25 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/UnhealthyDeploymentRestartTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/UnhealthyDeploymentRestartTest.java
@@ -68,8 +68,7 @@ public class UnhealthyDeploymentRestartTest {
configManager = new FlinkConfigManager(configuration);
flinkService = new TestingFlinkService(kubernetesClient);
context = flinkService.getContext();
- testController =
- new TestingFlinkDeploymentController(configManager, kubernetesClient, flinkService);
+ testController = new TestingFlinkDeploymentController(configManager, flinkService);
kubernetesClient.resource(TestUtils.buildApplicationCluster()).createOrReplace();
flinkService.setMetricValue(NUM_RESTARTS_METRIC_NAME, "0");
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
index f791ad25..4a9940bf 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/HealthProbeTest.java
@@ -182,31 +182,31 @@ public class HealthProbeTest {
var canaryManager =
new CanaryResourceManager<FlinkDeployment>(
- new FlinkConfigManager(new Configuration()), client);
+ new FlinkConfigManager(new Configuration()));
HealthProbe.INSTANCE.registerCanaryResourceManager(canaryManager);
// No canary resources
assertTrue(HealthProbe.INSTANCE.isHealthy());
var canary = TestUtils.createCanaryDeployment();
- canaryManager.handleCanaryResourceReconciliation(ReconciliationUtils.clone(canary));
+ canaryManager.handleCanaryResourceReconciliation(ReconciliationUtils.clone(canary), client);
// Canary resource healthy before health check
assertTrue(HealthProbe.INSTANCE.isHealthy());
// No reconciliation before health check
- canaryManager.checkHealth(ResourceID.fromResource(canary));
+ canaryManager.checkHealth(ResourceID.fromResource(canary), client);
assertFalse(HealthProbe.INSTANCE.isHealthy());
// Healthy again
canary.getMetadata().setGeneration(2L);
- canaryManager.handleCanaryResourceReconciliation(ReconciliationUtils.clone(canary));
- canaryManager.checkHealth(ResourceID.fromResource(canary));
+ canaryManager.handleCanaryResourceReconciliation(ReconciliationUtils.clone(canary), client);
+ canaryManager.checkHealth(ResourceID.fromResource(canary), client);
assertTrue(HealthProbe.INSTANCE.isHealthy());
canary.getMetadata().setGeneration(3L);
- canaryManager.handleCanaryResourceReconciliation(ReconciliationUtils.clone(canary));
- canaryManager.checkHealth(ResourceID.fromResource(canary));
+ canaryManager.handleCanaryResourceReconciliation(ReconciliationUtils.clone(canary), client);
+ canaryManager.checkHealth(ResourceID.fromResource(canary), client);
assertTrue(HealthProbe.INSTANCE.isHealthy());
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java
index 0df27576..6a99ccf8 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java
@@ -72,12 +72,12 @@ public class FlinkResourceListenerTest {
assertEquals(deployment, listener1.updates.get(0).getFlinkResource());
deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.ERROR);
- statusRecorder.patchAndCacheStatus(deployment);
+ statusRecorder.patchAndCacheStatus(deployment, kubernetesClient);
assertEquals(2, listener1.updates.size());
assertEquals(deployment, listener1.updates.get(1).getFlinkResource());
deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
- statusRecorder.patchAndCacheStatus(deployment);
+ statusRecorder.patchAndCacheStatus(deployment, kubernetesClient);
assertEquals(3, listener1.updates.size());
assertEquals(deployment, listener1.updates.get(2).getFlinkResource());
@@ -105,14 +105,16 @@ public class FlinkResourceListenerTest {
EventRecorder.Type.Warning,
EventRecorder.Reason.SavepointError,
EventRecorder.Component.Operator,
- "err");
+ "err",
+ kubernetesClient);
assertEquals(1, listener1.events.size());
eventRecorder.triggerEvent(
deployment,
EventRecorder.Type.Warning,
EventRecorder.Reason.SavepointError,
EventRecorder.Component.Operator,
- "err");
+ "err",
+ kubernetesClient);
assertEquals(2, listener1.events.size());
for (int i = 0; i < listener1.events.size(); i++) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
index 0bdb8dc3..68c0c227 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
@@ -72,13 +72,13 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class ApplicationObserverTest extends OperatorTestBase {
@Getter private KubernetesClient kubernetesClient;
- private final Context<FlinkDeployment> readyContext =
- TestUtils.createContextWithReadyJobManagerDeployment();
+ private Context<FlinkDeployment> readyContext;
private TestObserverAdapter<FlinkDeployment> observer;
@Override
public void setup() {
observer = new TestObserverAdapter<>(this, new ApplicationObserver(eventRecorder));
+ readyContext = TestUtils.createContextWithReadyJobManagerDeployment(kubernetesClient);
}
@Test
@@ -769,7 +769,8 @@ public class ApplicationObserverTest extends OperatorTestBase {
() ->
observer.observe(
deployment,
- TestUtils.createContextWithInProgressDeployment()));
+ TestUtils.createContextWithInProgressDeployment(
+ kubernetesClient)));
assertEquals(podFailedMessage, exception.getMessage());
}
@@ -778,7 +779,7 @@ public class ApplicationObserverTest extends OperatorTestBase {
var kubernetesDeployment = TestUtils.createDeployment(true);
kubernetesDeployment.getMetadata().setAnnotations(new HashMap<>());
- var context = TestUtils.createContextWithDeployment(kubernetesDeployment);
+ var context = TestUtils.createContextWithDeployment(kubernetesDeployment, kubernetesClient);
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
deployment.getMetadata().setGeneration(123L);
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 25e9e704..c08cd46e 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
@@ -46,7 +46,7 @@ public class SessionObserverTest extends OperatorTestBase {
@Getter private KubernetesClient kubernetesClient;
private final Context<FlinkDeployment> readyContext =
- TestUtils.createContextWithReadyJobManagerDeployment();
+ TestUtils.createContextWithReadyJobManagerDeployment(kubernetesClient);
private TestObserverAdapter<FlinkDeployment> observer;
@Override
@@ -106,7 +106,7 @@ public class SessionObserverTest extends OperatorTestBase {
var kubernetesDeployment = TestUtils.createDeployment(true);
kubernetesDeployment.getMetadata().setAnnotations(new HashMap<>());
- var context = TestUtils.createContextWithDeployment(kubernetesDeployment);
+ var context = TestUtils.createContextWithDeployment(kubernetesDeployment, kubernetesClient);
FlinkDeployment deployment = TestUtils.buildSessionCluster();
deployment.getMetadata().setGeneration(123L);
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverTest.java
index cd82f098..6de56566 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverTest.java
@@ -66,14 +66,13 @@ public class FlinkSessionJobObserverTest extends OperatorTestBase {
observer = new TestObserverAdapter<>(this, new FlinkSessionJobObserver(eventRecorder));
reconciler =
new TestReconcilerAdapter<>(
- this,
- new SessionJobReconciler(kubernetesClient, eventRecorder, statusRecorder));
+ this, new SessionJobReconciler(eventRecorder, statusRecorder));
}
@Test
public void testBasicObserve() throws Exception {
final var sessionJob = TestUtils.buildSessionJob();
- final var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ final var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
// observe the brand new job, nothing to do.
observer.observe(sessionJob, readyContext);
@@ -167,7 +166,7 @@ public class FlinkSessionJobObserverTest extends OperatorTestBase {
final var sessionJob = TestUtils.buildSessionJob();
final var readyContext =
TestUtils.createContextWithReadyFlinkDeployment(
- Map.of(RestOptions.PORT.key(), "8088"));
+ Map.of(RestOptions.PORT.key(), "8088"), kubernetesClient);
// submit job
reconciler.reconcile(sessionJob, readyContext);
@@ -187,7 +186,7 @@ public class FlinkSessionJobObserverTest extends OperatorTestBase {
@Test
public void testObserveSavepoint() throws Exception {
final var sessionJob = TestUtils.buildSessionJob();
- final var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ final var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
// submit job
reconciler.reconcile(sessionJob, readyContext);
var jobID = sessionJob.getStatus().getJobStatus().getJobId();
@@ -233,7 +232,7 @@ public class FlinkSessionJobObserverTest extends OperatorTestBase {
@Test
public void testObserveCheckpoint() throws Exception {
final var sessionJob = TestUtils.buildSessionJob();
- final var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ final var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
// submit job
reconciler.reconcile(sessionJob, readyContext);
var jobID = sessionJob.getStatus().getJobStatus().getJobId();
@@ -273,7 +272,7 @@ public class FlinkSessionJobObserverTest extends OperatorTestBase {
public void testObserveAlreadySubmitted() {
final var sessionJob = TestUtils.buildSessionJob();
sessionJob.getMetadata().setGeneration(10L);
- final var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ final var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
flinkService.setSessionJobSubmittedCallback(
() -> {
@@ -303,7 +302,7 @@ public class FlinkSessionJobObserverTest extends OperatorTestBase {
public void testObserveAlreadyUpgraded() throws Exception {
final var sessionJob = TestUtils.buildSessionJob();
sessionJob.getMetadata().setGeneration(10L);
- final var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ final var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
reconciler.reconcile(sessionJob, readyContext);
observer.observe(sessionJob, readyContext);
@@ -354,7 +353,7 @@ public class FlinkSessionJobObserverTest extends OperatorTestBase {
public void testOrphanedJob() throws Exception {
final var sessionJob = TestUtils.buildSessionJob();
sessionJob.getMetadata().setGeneration(10L);
- final var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ final var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
reconciler.reconcile(sessionJob, readyContext);
observer.observe(sessionJob, readyContext);
@@ -419,7 +418,8 @@ public class FlinkSessionJobObserverTest extends OperatorTestBase {
ReconciliationUtils.updateStatusBeforeDeploymentAttempt(sessionJob, new Configuration());
assertFalse(sessionJob.getStatus().getReconciliationStatus().isBeforeFirstDeployment());
- observer.observe(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ observer.observe(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertTrue(sessionJob.getStatus().getReconciliationStatus().isBeforeFirstDeployment());
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
index 3c272d04..d903cacc 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
@@ -135,10 +135,7 @@ public class ApplicationReconcilerTest extends OperatorTestBase {
public void setup() {
appReconciler =
new ApplicationReconciler(
- kubernetesClient,
- eventRecorder,
- statusRecorder,
- new NoopJobAutoscalerFactory());
+ eventRecorder, statusRecorder, new NoopJobAutoscalerFactory());
reconciler = new TestReconcilerAdapter<>(this, appReconciler);
operatorConfig = configManager.getOperatorConfiguration();
executorService = Executors.newDirectExecutorService();
@@ -154,13 +151,15 @@ public class ApplicationReconcilerTest extends OperatorTestBase {
FlinkDeployment deployment = TestUtils.buildApplicationCluster(flinkVersion);
// session ready
- reconciler.reconcile(deployment, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ deployment, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
verifyAndSetRunningJobsToStatus(deployment, flinkService.listJobs());
// clean up
assertEquals(
null, deployment.getStatus().getJobStatus().getSavepointInfo().getLastSavepoint());
- reconciler.cleanup(deployment, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ deployment, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(
"savepoint_0",
deployment
@@ -182,13 +181,15 @@ public class ApplicationReconcilerTest extends OperatorTestBase {
.put(KubernetesOperatorConfigOptions.SAVEPOINT_ON_DELETION.key(), "true");
// session ready
- reconciler.reconcile(deployment, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ deployment, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
verifyAndSetRunningJobsToStatus(deployment, flinkService.listJobs());
// clean up
assertEquals(
null, deployment.getStatus().getJobStatus().getSavepointInfo().getLastSavepoint());
- reconciler.cleanup(deployment, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ deployment, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(
"savepoint_0",
deployment
@@ -793,11 +794,7 @@ public class ApplicationReconcilerTest extends OperatorTestBase {
var ctxFactory =
new TestingFlinkResourceContextFactory(
- getKubernetesClient(),
- configManager,
- operatorMetricGroup,
- nativeService,
- eventRecorder);
+ configManager, operatorMetricGroup, nativeService, eventRecorder);
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
// Set all the properties required by the rescale api
@@ -879,14 +876,10 @@ public class ApplicationReconcilerTest extends OperatorTestBase {
public void testApplyAutoscalerParallelism() throws Exception {
var ctxFactory =
new TestingFlinkResourceContextFactory(
- getKubernetesClient(),
- configManager,
- operatorMetricGroup,
- flinkService,
- eventRecorder);
+ configManager, operatorMetricGroup, flinkService, eventRecorder);
var overrides = new HashMap<String, String>();
JobAutoScalerFactory autoscalerFactory =
- (k, r) ->
+ (r) ->
new NoopJobAutoscalerFactory() {
@Override
public Map<String, String> getParallelismOverrides(
@@ -900,9 +893,7 @@ public class ApplicationReconcilerTest extends OperatorTestBase {
}
};
- appReconciler =
- new ApplicationReconciler(
- kubernetesClient, eventRecorder, statusRecorder, autoscalerFactory);
+ appReconciler = new ApplicationReconciler(eventRecorder, statusRecorder, autoscalerFactory);
var deployment = TestUtils.buildApplicationCluster();
appReconciler.reconcile(ctxFactory.getResourceContext(deployment, context));
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java
index 1d7ed2ca..fcde70b6 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java
@@ -78,10 +78,7 @@ public class ApplicationReconcilerUpgradeModeTest extends OperatorTestBase {
new TestReconcilerAdapter<>(
this,
new ApplicationReconciler(
- kubernetesClient,
- eventRecorder,
- statusRecorder,
- new NoopJobAutoscalerFactory()));
+ eventRecorder, statusRecorder, new NoopJobAutoscalerFactory()));
}
@ParameterizedTest
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 a087e6e2..87f44b49 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
@@ -60,8 +60,7 @@ public class SessionReconcilerTest extends OperatorTestBase {
public void setup() {
reconciler =
new TestReconcilerAdapter<>(
- this,
- new SessionReconciler(kubernetesClient, eventRecorder, statusRecorder));
+ this, new SessionReconciler(eventRecorder, statusRecorder));
}
@Test
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
index 18f8dce7..ab518429 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
@@ -92,8 +92,7 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
configManager = new FlinkConfigManager(configuration);
reconciler =
new TestReconcilerAdapter<>(
- this,
- new SessionJobReconciler(kubernetesClient, eventRecorder, statusRecorder));
+ this, new SessionJobReconciler(eventRecorder, statusRecorder));
}
@Test
@@ -105,13 +104,15 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
// session ready
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
// clean up
- reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(
"savepoint_0",
sessionJob
@@ -131,13 +132,15 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
.put(KubernetesOperatorConfigOptions.SAVEPOINT_ON_DELETION.key(), "true");
// session ready
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
// clean up
- reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(
"savepoint_0",
sessionJob
@@ -161,12 +164,14 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
assertEquals(0, flinkService.listJobs().size());
// session ready
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
// clean up
- reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(FINISHED, flinkService.listJobs().get(0).f1.getJobState());
}
@@ -175,20 +180,25 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
// session ready
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
// clean up
flinkService.setPortReady(false);
var deleteControl =
- reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ sessionJob,
+ TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(10_000, deleteControl.getScheduleDelay().get());
assertEquals(RUNNING, flinkService.listJobs().get(0).f1.getJobState());
flinkService.setPortReady(true);
deleteControl =
- reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ sessionJob,
+ TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(true, deleteControl.isRemoveFinalizer());
assertEquals(FINISHED, flinkService.listJobs().get(0).f1.getJobState());
}
@@ -198,17 +208,22 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
// session ready
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
// clean up
flinkService.setFlinkJobNotFound(true);
var deleteControl =
- reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ sessionJob,
+ TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
deleteControl =
- reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ sessionJob,
+ TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(true, deleteControl.isRemoveFinalizer());
}
@@ -217,17 +232,22 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
// session ready
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
// clean up
flinkService.setFlinkJobTerminatedWithoutCancellation(true);
var deleteControl =
- reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ sessionJob,
+ TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
deleteControl =
- reconciler.cleanup(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.cleanup(
+ sessionJob,
+ TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(true, deleteControl.isRemoveFinalizer());
}
@@ -236,14 +256,17 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
// session ready
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
sessionJob.getSpec().setRestartNonce(2L);
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
assertEquals(FINISHED, flinkService.listJobs().get(0).f1.getJobState());
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
}
@@ -251,7 +274,7 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
@Test
public void testRestartWhenFailed() throws Exception {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
- var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
// session ready
reconciler.reconcile(sessionJob, readyContext);
@@ -271,7 +294,8 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
var initSavepointPath = "file:///init-sp";
sessionJob.getSpec().getJob().setInitialSavepointPath(initSavepointPath);
- reconciler.reconcile(sessionJob, TestUtils.createContextWithReadyFlinkDeployment());
+ reconciler.reconcile(
+ sessionJob, TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient));
verifyAndSetRunningJobsToStatus(
sessionJob,
JobState.RUNNING,
@@ -284,7 +308,7 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
public void testStatelessUpgrade() throws Exception {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
- var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
reconciler.reconcile(sessionJob, readyContext);
assertEquals(1, flinkService.listJobs().size());
verifyAndSetRunningJobsToStatus(
@@ -313,7 +337,7 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
public void testSavepointUpgrade() throws Exception {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
- var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
reconciler.reconcile(sessionJob, readyContext);
// start the job
assertEquals(1, flinkService.listJobs().size());
@@ -376,7 +400,7 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
assertFalse(SnapshotUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
- var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
reconciler.reconcile(sessionJob, readyContext);
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
@@ -497,7 +521,7 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
assertFalse(SnapshotUtils.checkpointInProgress(getJobStatus(sessionJob)));
- var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
reconciler.reconcile(sessionJob, readyContext);
verifyAndSetRunningJobsToStatus(
sessionJob, JobState.RUNNING, RECONCILING.name(), null, flinkService.listJobs());
@@ -574,7 +598,7 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
throws Exception {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
- var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
reconciler.reconcile(sessionJob, readyContext);
assertEquals(1, flinkService.listJobs().size());
@@ -644,7 +668,7 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
- var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
reconciler.reconcile(sessionJob, readyContext);
assertEquals(1, flinkService.listJobs().size());
@@ -735,7 +759,7 @@ public class SessionJobReconcilerTest extends OperatorTestBase {
@Test
public void testJobUpgradeIgnorePendingSavepoint() throws Exception {
- var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ var readyContext = TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
reconciler.reconcile(sessionJob, readyContext);
verifyAndSetRunningJobsToStatus(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java
index 1d2e2b4c..fb823500 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/NativeFlinkServiceTest.java
@@ -97,7 +97,7 @@ public class NativeFlinkServiceTest {
configuration.set(KubernetesConfigOptions.CLUSTER_ID, TestUtils.TEST_DEPLOYMENT_NAME);
configuration.set(KubernetesConfigOptions.NAMESPACE, TestUtils.TEST_NAMESPACE);
configuration.set(FLINK_VERSION, FlinkVersion.v1_15);
- eventRecorder = new EventRecorder(client, eventCollector);
+ eventRecorder = new EventRecorder(eventCollector);
operatorConfig = FlinkOperatorConfiguration.fromConfiguration(configuration);
executorService = Executors.newDirectExecutorService();
}
@@ -247,7 +247,7 @@ public class NativeFlinkServiceTest {
service.scale(
new FlinkDeploymentContext(
flinkDep,
- TestUtils.createEmptyContext(),
+ TestUtils.createEmptyContextWithClient(client),
null,
configManager,
ctx -> service),
@@ -445,7 +445,7 @@ public class NativeFlinkServiceTest {
service.scale(
new FlinkDeploymentContext(
depCopy,
- TestUtils.createEmptyContext(),
+ TestUtils.createEmptyContextWithClient(client),
null,
configManager,
c -> service),
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
index b86f643d..b7ad3b3a 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkServiceTest.java
@@ -147,7 +147,6 @@ public class StandaloneFlinkServiceTest {
flinkDeployment.getSpec().getTaskManager().setReplicas(5);
var ctx =
new FlinkResourceContextFactory(
- kubernetesClient,
configManager,
TestUtils.createTestMetricGroup(new Configuration()),
null)
@@ -176,7 +175,6 @@ public class StandaloneFlinkServiceTest {
.serializeAndSetLastReconciledSpec(flinkDeployment.getSpec(), flinkDeployment);
ctx =
new FlinkResourceContextFactory(
- kubernetesClient,
configManager,
TestUtils.createTestMetricGroup(new Configuration()),
null)
@@ -209,7 +207,6 @@ public class StandaloneFlinkServiceTest {
var ctx =
new FlinkResourceContextFactory(
- kubernetesClient,
configManager,
TestUtils.createTestMetricGroup(new Configuration()),
null)
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
index b43d8723..aab8eb80 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
@@ -41,23 +41,23 @@ public class StatusRecorderTest {
public void testPatchOnlyWhenChanged() throws InterruptedException {
var helper =
new StatusRecorder<FlinkDeployment, FlinkDeploymentStatus>(
- kubernetesClient, new MetricManager<>(), (e, s) -> {});
+ new MetricManager<>(), (e, s) -> {});
var deployment = TestUtils.buildApplicationCluster();
kubernetesClient.resource(deployment).createOrReplace();
var lastRequest = mockServer.getLastRequest();
- helper.patchAndCacheStatus(deployment);
+ helper.patchAndCacheStatus(deployment, kubernetesClient);
assertTrue(mockServer.getLastRequest() != lastRequest);
lastRequest = mockServer.getLastRequest();
deployment.getStatus().getReconciliationStatus().setState(ReconciliationState.ROLLING_BACK);
- helper.patchAndCacheStatus(deployment);
+ helper.patchAndCacheStatus(deployment, kubernetesClient);
// We intentionally compare references
assertTrue(mockServer.getLastRequest() != lastRequest);
lastRequest = mockServer.getLastRequest();
// No update
- helper.patchAndCacheStatus(deployment);
+ helper.patchAndCacheStatus(deployment, kubernetesClient);
assertTrue(mockServer.getLastRequest() == lastRequest);
}
}