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 5bcc06f60 [FLINK-41050] Canary health checks revert canary metadata
changes and restart a healthy operator (#1227)
5bcc06f60 is described below
commit 5bcc06f60b5bc5a0d510cf19e413a1d8cd9938a5
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Thu Oct 29 09:27:01 2026 +0200
[FLINK-41050] Canary health checks revert canary metadata changes and
restart a healthy operator (#1227)
---
.../flink/kubernetes/operator/FlinkOperator.java | 4 +
.../operator/health/CanaryResourceManager.java | 159 +++++++++---
.../operator/health/CanaryResourceManagerTest.java | 276 +++++++++++++++++++++
.../operator/health/HealthProbeTest.java | 18 +-
4 files changed, 411 insertions(+), 46 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index b37aefdfd..8a5af0e6b 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
@@ -85,6 +85,7 @@ public class FlinkOperator {
private final FlinkConfigManager configManager;
private final Set<FlinkResourceValidator> validators;
@VisibleForTesting final Set<RegisteredController<?>>
registeredControllers = new HashSet<>();
+ private final Set<CanaryResourceManager<?>> canaryResourceManagers = new
HashSet<>();
private final KubernetesOperatorMetricGroup metricGroup;
private final Collection<FlinkResourceListener> listeners;
private final OperatorHealthService operatorHealthService;
@@ -147,6 +148,7 @@ public class FlinkOperator {
controller.changeNamespaces(namespaces);
}
});
+ canaryResourceManagers.forEach(manager ->
manager.retainNamespaces(namespaces));
}
private void overrideOperatorConfigs(ConfigurationServiceOverrider
overrider) {
@@ -198,6 +200,7 @@ public class FlinkOperator {
var observerFactory = new
FlinkDeploymentObserverFactory(eventRecorder);
var canaryResourceManager = new
CanaryResourceManager<FlinkDeployment>(configManager);
HealthProbe.INSTANCE.registerCanaryResourceManager(canaryResourceManager);
+ canaryResourceManagers.add(canaryResourceManager);
var controller =
new FlinkDeploymentController(
@@ -222,6 +225,7 @@ public class FlinkOperator {
var observer = new FlinkSessionJobObserver(eventRecorder);
var canaryResourceManager = new
CanaryResourceManager<FlinkSessionJob>(configManager);
HealthProbe.INSTANCE.registerCanaryResourceManager(canaryResourceManager);
+ canaryResourceManagers.add(canaryResourceManager);
var controller =
new FlinkSessionJobController(
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 f141239e0..e3eccc0b1 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
@@ -21,21 +21,30 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
-import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.javaoperatorsdk.operator.api.reconciler.Constants;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-
-/** Logic encapsulating canary tests. */
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Tracks the canary resources of one kind, labeled with {@link
#CANARY_LABEL}, to verify that the
+ * operator still picks up and reconciles spec changes. Each canary spec is
periodically updated by
+ * bumping its restart nonce, and the canary is reported unhealthy when the
update is not reconciled
+ * within the canary resource timeout, or when the update itself keeps
failing. Canaries are never
+ * deployed, their reconciliation stops at this class.
+ */
@RequiredArgsConstructor
public class CanaryResourceManager<CR extends AbstractFlinkResource<?, ?>> {
@@ -43,6 +52,8 @@ public class CanaryResourceManager<CR extends
AbstractFlinkResource<?, ?>> {
public static final String CANARY_LABEL = "flink.apache.org/canary";
+ @VisibleForTesting static final int MAX_FAILED_UPDATES = 3;
+
private final ConcurrentHashMap<ResourceID, CanaryResourceState>
canaryResources =
new ConcurrentHashMap<>();
@@ -55,28 +66,36 @@ public class CanaryResourceManager<CR extends
AbstractFlinkResource<?, ?>> {
}
public boolean handleCanaryResourceReconciliation(CR resource,
KubernetesClient client) {
+ var resourceId = ResourceID.fromResource(resource);
if (!isCanaryResource(resource)) {
+ if (canaryResources.remove(resourceId) != null) {
+ LOG.info(
+ "Reconciled {} without the canary label. Stopping
health checks",
+ resourceId);
+ }
return false;
}
- var resourceId = ResourceID.fromResource(resource);
-
LOG.info("Reconciling canary resource");
- canaryResources.compute(
- resourceId,
- (id, previousState) -> {
- boolean firstReconcile = false;
- if (previousState == null) {
- firstReconcile = true;
- previousState = new CanaryResourceState();
- }
- previousState.onReconcile(resource);
- if (firstReconcile) {
- updateSpecAndScheduleHealthCheck(resourceId,
previousState, client);
- }
- return previousState;
- });
+ var firstReconcile = new AtomicBoolean(false);
+ var state =
+ canaryResources.compute(
+ resourceId,
+ (id, previousState) -> {
+ if (previousState == null) {
+ firstReconcile.set(true);
+ previousState = new CanaryResourceState();
+ previousState.bumpedGeneration =
+ resource.getMetadata().getGeneration();
+ }
+ previousState.onReconcile(resource);
+ return previousState;
+ });
+
+ if (firstReconcile.get()) {
+ updateSpecAndScheduleHealthCheck(resourceId, state, client);
+ }
return true;
}
@@ -92,6 +111,23 @@ public class CanaryResourceManager<CR extends
AbstractFlinkResource<?, ?>> {
return true;
}
+ public void retainNamespaces(Set<String> namespaces) {
+ // The operator switched to watching all namespaces
+ if (namespaces.contains(Constants.WATCH_ALL_NAMESPACES)) {
+ return;
+ }
+
+ for (var resourceId : canaryResources.keySet()) {
+ var watched =
resourceId.getNamespace().map(namespaces::contains).orElse(true);
+ if (!watched && canaryResources.remove(resourceId) != null) {
+ LOG.info(
+ "Namespace of canary resource {} is no longer watched.
Stopping health checks",
+ resourceId);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
private void updateSpecAndScheduleHealthCheck(
ResourceID resourceID, CanaryResourceState crs, KubernetesClient
client) {
var canaryTimeout =
@@ -99,18 +135,59 @@ public class CanaryResourceManager<CR extends
AbstractFlinkResource<?, ?>> {
.getDefaultConfig()
.get(KubernetesOperatorConfigOptions.CANARY_RESOURCE_TIMEOUT);
- Long restartNonce = crs.resource.getSpec().getRestartNonce();
- crs.resource.getSpec().setRestartNonce(restartNonce == null ? 1L :
restartNonce + 1);
- crs.previousGeneration = crs.resource.getMetadata().getGeneration();
+ Long newGeneration = null;
+ var deleted = false;
+ var labelRemoved = false;
+ try {
+ var canaryResource =
+ client.resources((Class<CR>) crs.resource.getClass())
+
.inNamespace(resourceID.getNamespace().orElse(null))
+ .withName(resourceID.getName());
+ var latest = canaryResource.get();
+ if (latest == null) {
+ deleted = true;
+ } else if (!isCanaryResource(latest)) {
+ labelRemoved = true;
+ } else {
+ var restartNonce = latest.getSpec().getRestartNonce();
+ latest.getSpec().setRestartNonce(restartNonce == null ? 1L :
restartNonce + 1);
+ newGeneration =
client.resource(latest).update().getMetadata().getGeneration();
+ }
+ } catch (KubernetesClientException e) {
+ if (e.getCode() == 404) {
+ deleted = true;
+ } else {
+ LOG.warn("Could not bump canary resource {}, will retry",
resourceID, e);
+ }
+ } catch (Exception e) {
+ LOG.warn("Could not bump canary resource {}, will retry",
resourceID, e);
+ }
- LOG.info("Scheduling canary check for {} in {}s", resourceID,
canaryTimeout.toSeconds());
+ if (deleted) {
+ LOG.info(
+ "Canary resource {} was deleted without cleanup. Stopping
health checks",
+ resourceID);
+ canaryResources.remove(resourceID, crs);
+ return;
+ }
- try {
- client.resource(ReconciliationUtils.clone(crs.resource)).replace();
- } catch (Throwable t) {
- LOG.warn("Could not bump canary deployment, it may have been
deleted", t);
+ if (labelRemoved) {
+ LOG.info(
+ "Canary label removed from {}, skipping the spec bump.
Stopping health checks",
+ resourceID);
+ canaryResources.remove(resourceID, crs);
+ return;
}
+ if (newGeneration == null) {
+ crs.failedBumps++;
+ } else {
+ crs.bumpedGeneration = newGeneration;
+ crs.failedBumps = 0;
+ }
+
+ LOG.info("Scheduling canary check for {} in {}s", resourceID,
canaryTimeout.toSeconds());
+
executorService.schedule(
() -> checkHealth(resourceID, client),
canaryTimeout.toMillis(),
@@ -125,20 +202,31 @@ public class CanaryResourceManager<CR extends
AbstractFlinkResource<?, ?>> {
return;
}
- // We did not reconcile since the last spec update within deadline
- if (crs.canaryReconciledSinceUpdate()) {
+ if (crs.failedBumps >= MAX_FAILED_UPDATES) {
+ LOG.error(
+ "Canary deployment {} spec could not be updated in {}
consecutive attempts",
+ resourceID,
+ crs.failedBumps);
+ crs.isHealthy = false;
+ } else if (crs.canaryReconciledSinceUpdate() && crs.failedBumps > 0) {
+ LOG.warn(
+ "Canary deployment {} healthy, verified against its last
successful spec update. The last {} spec update attempts failed",
+ resourceID,
+ crs.failedBumps);
+ crs.isHealthy = true;
+ } else if (crs.canaryReconciledSinceUpdate()) {
LOG.info("Canary deployment healthy");
crs.isHealthy = true;
} else {
+ // We did not reconcile since the last spec update within deadline
LOG.error(
- "Canary deployment {} latest spec not reconciled. Expected
generation larger than {}, received {}",
+ "Canary deployment {} latest spec not reconciled. Expected
generation at least {}, received {}",
resourceID,
- crs.previousGeneration,
+ crs.bumpedGeneration,
crs.resource.getMetadata().getGeneration());
crs.isHealthy = false;
}
- // Update spec and reschedule health check
updateSpecAndScheduleHealthCheck(resourceID, crs, client);
}
@@ -157,9 +245,8 @@ public class CanaryResourceManager<CR extends
AbstractFlinkResource<?, ?>> {
private class CanaryResourceState {
CR resource;
-
- long previousGeneration;
-
+ long bumpedGeneration;
+ int failedBumps;
boolean isHealthy = true;
void onReconcile(CR cr) {
@@ -167,7 +254,7 @@ public class CanaryResourceManager<CR extends
AbstractFlinkResource<?, ?>> {
}
boolean canaryReconciledSinceUpdate() {
- return resource.getMetadata().getGeneration() > previousGeneration;
+ return resource.getMetadata().getGeneration() >= bumpedGeneration;
}
}
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManagerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManagerTest.java
new file mode 100644
index 000000000..5d55f3ab6
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/health/CanaryResourceManagerTest.java
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.health;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.TestUtils;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+
+import io.fabric8.kubernetes.api.model.GenericKubernetesResourceBuilder;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.ConfigBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientBuilder;
+import io.fabric8.kubernetes.client.dsl.base.ResourceDefinitionContext;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import io.javaoperatorsdk.operator.api.reconciler.Constants;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Tests for {@link CanaryResourceManager}. */
+@EnableKubernetesMockClient(crud = true)
+public class CanaryResourceManagerTest {
+
+ KubernetesClient client;
+ KubernetesMockServer mockServer;
+
+ private CanaryResourceManager<FlinkDeployment> canaryManager;
+ private ResourceID canaryId;
+
+ @BeforeAll
+ public static void setAutoTryKubeConfig() {
+
System.setProperty(Config.KUBERNETES_AUTH_TRYKUBECONFIG_SYSTEM_PROPERTY,
"false");
+ }
+
+ @BeforeEach
+ public void setup() {
+ canaryManager = new CanaryResourceManager<>(new FlinkConfigManager(new
Configuration()));
+ var canary = TestUtils.createCanaryDeployment();
+ client.resource(canary).create();
+ canaryId = ResourceID.fromResource(canary);
+ }
+
+ @Test
+ public void testHealthDependsOnReconcilingTheBump() {
+ reconcileLatest();
+ assertEquals(1, canaryManager.getNumberOfActiveCanaries());
+ assertTrue(canaryManager.allCanariesHealthy());
+
+ canaryManager.checkHealth(canaryId, client);
+ assertFalse(canaryManager.allCanariesHealthy());
+
+ reconcileLatest();
+ canaryManager.checkHealth(canaryId, client);
+ assertTrue(canaryManager.allCanariesHealthy());
+ }
+
+ @Test
+ public void testBumpPreservesConcurrentMetadataChanges() {
+ reconcileLatest();
+ var nonce = latest().getSpec().getRestartNonce();
+
+ client.resource(latest())
+ .edit(
+ cr -> {
+ var labels = new
HashMap<>(cr.getMetadata().getLabels());
+ labels.put("test", "label");
+ cr.getMetadata().setLabels(labels);
+ return cr;
+ });
+
+ canaryManager.checkHealth(canaryId, client);
+
+ var canary = latest();
+ assertEquals("label", canary.getMetadata().getLabels().get("test"));
+ assertEquals(nonce + 1, canary.getSpec().getRestartNonce());
+ }
+
+ @Test
+ public void testCanaryWithoutSpec() throws InterruptedException {
+ var context =
+ new ResourceDefinitionContext.Builder()
+ .withGroup("flink.apache.org")
+ .withVersion("v1beta1")
+ .withKind("FlinkDeployment")
+ .withPlural("flinkdeployments")
+ .withNamespaced(true)
+ .build();
+ client.genericKubernetesResources(context)
+ .inNamespace("default")
+ .resource(
+ new GenericKubernetesResourceBuilder()
+ .withApiVersion("flink.apache.org/v1beta1")
+ .withKind("FlinkDeployment")
+ .withNewMetadata()
+ .withName("no-spec")
+ .withNamespace("default")
+
.withLabels(Map.of(CanaryResourceManager.CANARY_LABEL, "true"))
+ .endMetadata()
+ .build())
+ .create();
+ var noSpec =
+
client.resources(FlinkDeployment.class).inNamespace("default").withName("no-spec");
+
+
assertTrue(canaryManager.handleCanaryResourceReconciliation(noSpec.get(),
client));
+
+ var lastRequest = mockServer.getLastRequest();
+ assertEquals("PUT", lastRequest.getMethod());
+ assertEquals(1L, noSpec.get().getSpec().getRestartNonce());
+ }
+
+ @Test
+ public void testFailedBumpsReportUnhealthyAfterMaxAttempts() {
+ reconcileLatest();
+ reconcileLatest();
+ try (var unreachableClient = unreachableClient()) {
+ canaryManager.checkHealth(canaryId, unreachableClient);
+ assertTrue(canaryManager.allCanariesHealthy());
+
+ for (int i = 1; i < CanaryResourceManager.MAX_FAILED_UPDATES; i++)
{
+ canaryManager.checkHealth(canaryId, unreachableClient);
+ assertTrue(canaryManager.allCanariesHealthy());
+ }
+
+ canaryManager.checkHealth(canaryId, unreachableClient);
+ assertFalse(canaryManager.allCanariesHealthy());
+ assertEquals(1, canaryManager.getNumberOfActiveCanaries());
+ }
+
+ canaryManager.checkHealth(canaryId, client);
+ reconcileLatest();
+ canaryManager.checkHealth(canaryId, client);
+ assertTrue(canaryManager.allCanariesHealthy());
+ }
+
+ @Test
+ public void testLateReconcileRecoversWhileBumpsFail() {
+ reconcileLatest();
+ try (var unreachableClient = unreachableClient()) {
+ canaryManager.checkHealth(canaryId, unreachableClient);
+ assertFalse(canaryManager.allCanariesHealthy());
+
+ reconcileLatest();
+ canaryManager.checkHealth(canaryId, unreachableClient);
+ assertTrue(canaryManager.allCanariesHealthy());
+ }
+ }
+
+ @Test
+ public void testFailedFirstBump() {
+ try (var unreachableClient = unreachableClient()) {
+ assertTrue(
+ canaryManager.handleCanaryResourceReconciliation(latest(),
unreachableClient));
+ canaryManager.checkHealth(canaryId, unreachableClient);
+ assertTrue(canaryManager.allCanariesHealthy());
+ }
+ }
+
+ @Test
+ public void testDeletedWithoutCleanupStopsHealthChecks() {
+ reconcileLatest();
+
+ client.resource(latest()).delete();
+
+ canaryManager.checkHealth(canaryId, client);
+ assertEquals(0, canaryManager.getNumberOfActiveCanaries());
+ assertTrue(canaryManager.allCanariesHealthy());
+ }
+
+ @Test
+ public void testCanaryLabelRemoved() {
+ reconcileLatest();
+ var nonce = latest().getSpec().getRestartNonce();
+
+ client.resource(latest())
+ .edit(
+ cr -> {
+
cr.getMetadata().getLabels().remove(CanaryResourceManager.CANARY_LABEL);
+ return cr;
+ });
+
+ canaryManager.checkHealth(canaryId, client);
+ assertEquals(0, canaryManager.getNumberOfActiveCanaries());
+ assertEquals(nonce, latest().getSpec().getRestartNonce());
+ }
+
+ @Test
+ public void testNamespaceNoLongerWatched() {
+ var otherCanary = TestUtils.createCanaryDeployment();
+ otherCanary.getMetadata().setNamespace("other");
+ client.resource(otherCanary).create();
+ reconcileLatest();
+
assertTrue(canaryManager.handleCanaryResourceReconciliation(otherCanary,
client));
+ assertEquals(2, canaryManager.getNumberOfActiveCanaries());
+
+ canaryManager.retainNamespaces(Set.of(Constants.WATCH_ALL_NAMESPACES));
+ assertEquals(2, canaryManager.getNumberOfActiveCanaries());
+
+ var nonce = latest().getSpec().getRestartNonce();
+ canaryManager.retainNamespaces(Set.of("other"));
+ assertEquals(1, canaryManager.getNumberOfActiveCanaries());
+
+ canaryManager.checkHealth(canaryId, client);
+ assertEquals(1, canaryManager.getNumberOfActiveCanaries());
+ assertEquals(nonce, latest().getSpec().getRestartNonce());
+ assertTrue(canaryManager.allCanariesHealthy());
+ }
+
+ @Test
+ public void testReconcilingFormerCanaryStopsTracking() {
+ reconcileLatest();
+ assertEquals(1, canaryManager.getNumberOfActiveCanaries());
+
+ var formerCanary = latest();
+
formerCanary.getMetadata().getLabels().remove(CanaryResourceManager.CANARY_LABEL);
+
assertFalse(canaryManager.handleCanaryResourceReconciliation(formerCanary,
client));
+ assertEquals(0, canaryManager.getNumberOfActiveCanaries());
+ }
+
+ @Test
+ public void testDeletion() {
+ reconcileLatest();
+ assertTrue(canaryManager.handleCanaryResourceDeletion(latest()));
+ assertEquals(0, canaryManager.getNumberOfActiveCanaries());
+
+ canaryManager.checkHealth(canaryId, client);
+ assertEquals(0, canaryManager.getNumberOfActiveCanaries());
+ }
+
+ private void reconcileLatest() {
+ assertTrue(canaryManager.handleCanaryResourceReconciliation(latest(),
client));
+ }
+
+ private static KubernetesClient unreachableClient() {
+ return new KubernetesClientBuilder()
+ .withConfig(
+ new ConfigBuilder()
+ .withMasterUrl("http://localhost:1";)
+ .withRequestRetryBackoffLimit(0)
+ .build())
+ .build();
+ }
+
+ private FlinkDeployment latest() {
+ return client.resources(FlinkDeployment.class)
+ .inNamespace(canaryId.getNamespace().orElseThrow())
+ .withName(canaryId.getName())
+ .get();
+ }
+}
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 6ce4f16e2..654476b1b 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
@@ -23,7 +23,6 @@ import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
-import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.util.NetUtils;
import io.fabric8.kubernetes.client.Config;
@@ -188,25 +187,24 @@ public class HealthProbeTest {
// No canary resources
assertTrue(HealthProbe.INSTANCE.isHealthy());
- var canary = TestUtils.createCanaryDeployment();
-
canaryManager.handleCanaryResourceReconciliation(ReconciliationUtils.clone(canary),
client);
+ var canary =
client.resource(TestUtils.createCanaryDeployment()).create();
+ var canaryId = ResourceID.fromResource(canary);
+
canaryManager.handleCanaryResourceReconciliation(client.resource(canary).get(),
client);
// Canary resource healthy before health check
assertTrue(HealthProbe.INSTANCE.isHealthy());
// No reconciliation before health check
- canaryManager.checkHealth(ResourceID.fromResource(canary), client);
+ canaryManager.checkHealth(canaryId, client);
assertFalse(HealthProbe.INSTANCE.isHealthy());
// Healthy again
- canary.getMetadata().setGeneration(2L);
-
canaryManager.handleCanaryResourceReconciliation(ReconciliationUtils.clone(canary),
client);
- canaryManager.checkHealth(ResourceID.fromResource(canary), client);
+
canaryManager.handleCanaryResourceReconciliation(client.resource(canary).get(),
client);
+ canaryManager.checkHealth(canaryId, client);
assertTrue(HealthProbe.INSTANCE.isHealthy());
- canary.getMetadata().setGeneration(3L);
-
canaryManager.handleCanaryResourceReconciliation(ReconciliationUtils.clone(canary),
client);
- canaryManager.checkHealth(ResourceID.fromResource(canary), client);
+
canaryManager.handleCanaryResourceReconciliation(client.resource(canary).get(),
client);
+ canaryManager.checkHealth(canaryId, client);
assertTrue(HealthProbe.INSTANCE.isHealthy());
}
