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 8e567a5d [FLINK-39953] Fix cluster deletion timeout silently swallowed
during upgrade (#1138)
8e567a5d is described below
commit 8e567a5da7b22ebe95af4bfd1c36ee40cbba0b30
Author: Lucas Gameiro <[email protected]>
AuthorDate: Tue Aug 4 10:48:26 2026 +0200
[FLINK-39953] Fix cluster deletion timeout silently swallowed during
upgrade (#1138)
---
.../operator/service/AbstractFlinkService.java | 4 +
.../operator/service/NativeFlinkService.java | 42 ++++---
.../operator/service/AbstractFlinkServiceTest.java | 71 ++++++++++--
.../operator/service/NativeFlinkServiceTest.java | 64 ++++++++++
.../service/StandaloneFlinkServiceTest.java | 129 ++++++++++++++++++++-
5 files changed, 285 insertions(+), 25 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index 543b937f..1d48d10d 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -117,6 +117,7 @@ import io.fabric8.kubernetes.api.model.PodList;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.KubernetesClientTimeoutException;
import io.fabric8.kubernetes.client.dsl.Resource;
import io.fabric8.kubernetes.client.dsl.Waitable;
import lombok.SneakyThrows;
@@ -1242,6 +1243,9 @@ public abstract class AbstractFlinkService implements
FlinkService {
deleted.waitUntilCondition(
Objects::isNull, timeout.toMillis(),
TimeUnit.MILLISECONDS);
LOG.info("Completed {}", operation);
+ } catch (KubernetesClientTimeoutException e) {
+ // Rethrow timeouts before the broader catch swallows it
+ throw e;
} catch (KubernetesClientException kce) {
// We completely ignore not found errors and simply log others
if (kce.getCode() != HttpURLConnection.HTTP_NOT_FOUND) {
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 a5a63f8b..65c2c7cd 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
@@ -56,6 +56,7 @@ import io.fabric8.kubernetes.api.model.PodList;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientTimeoutException;
import io.fabric8.kubernetes.client.dsl.EditReplacePatchable;
import io.fabric8.kubernetes.client.dsl.base.PatchContext;
import io.fabric8.kubernetes.client.dsl.base.PatchType;
@@ -323,23 +324,30 @@ public class NativeFlinkService extends
AbstractFlinkService {
// here is to initiate JM shutdown before the TMs
var jmShutdownTimeout =
ObjectUtils.min(JM_SHUTDOWN_MAX_WAIT,
remainingTimeout.dividedBy(2));
- var remaining =
- deleteBlocking(
- "Scaling JobManager Deployment to zero",
- () -> {
- try {
- jmDeployment.patch(
- PatchContext.of(PatchType.JSON_MERGE),
SCALE_TO_ZERO);
- } catch (Exception ignore) {
- // Ignore all errors here as this is an
optional step
- return null;
- }
- return kubernetesClient
- .pods()
- .inNamespace(namespace)
-
.withLabels(KubernetesUtils.getJobManagerSelectors(clusterId));
- },
- jmShutdownTimeout);
+ Duration remaining;
+ try {
+ remaining =
+ deleteBlocking(
+ "Scaling JobManager Deployment to zero",
+ () -> {
+ try {
+ jmDeployment.patch(
+
PatchContext.of(PatchType.JSON_MERGE), SCALE_TO_ZERO);
+ } catch (Exception ignore) {
+ // Ignore all errors here as this is an
optional step
+ return null;
+ }
+ return kubernetesClient
+ .pods()
+ .inNamespace(namespace)
+ .withLabels(
+
KubernetesUtils.getJobManagerSelectors(clusterId));
+ },
+ jmShutdownTimeout);
+ } catch (KubernetesClientTimeoutException e) {
+ LOG.warn("Timed out waiting for JobManager scale-to-zero (optional
step), proceeding");
+ remaining = Duration.ZERO;
+ }
return remainingTimeout.minus(jmShutdownTimeout).plus(remaining);
}
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
index 19d50587..ebfe0616 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
@@ -110,6 +110,7 @@ import io.fabric8.kubernetes.api.model.apps.DeploymentList;
import io.fabric8.kubernetes.api.model.apps.DeploymentListBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.KubernetesClientTimeoutException;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import lombok.SneakyThrows;
@@ -1319,19 +1320,75 @@ public class AbstractFlinkServiceTest {
assertTrue(remainingMillis > 0);
assertTrue(remainingMillis < 10000 - deleteDelay / 2);
- // Test actual timeout
- remainingMillis =
- flinkService
- .deleteDeploymentBlocking(
+ // Timeout waiting for deletion should throw, not silently return
+ Assertions.assertThrows(
+ KubernetesClientTimeoutException.class,
+ () ->
+ flinkService.deleteDeploymentBlocking(
"Test",
client.apps()
.deployments()
.inNamespace(namespace)
.withName(deploymentName),
DeletionPropagation.BACKGROUND,
- Duration.ofMillis(10))
- .toMillis();
- assertEquals(0, remainingMillis);
+ Duration.ofMillis(10)));
+ }
+
+ @Test
+ public void testDeleteBlockingTimeoutThrows() {
+ String deploymentName = "test-cluster";
+ String namespace = "test-namespace";
+ String getUrl =
+ String.format(
+
"/apis/apps/v1/namespaces/%s/deployments?fieldSelector=metadata.name%%3D%s",
+ namespace, deploymentName);
+ String watchUrl =
+ String.format(
+
"/apis/apps/v1/namespaces/%s/deployments?allowWatchBookmarks=true&fieldSelector=metadata.name%%3D%s&timeoutSeconds=600&watch=true",
+ namespace, deploymentName);
+
+ Deployment deployment =
+ new DeploymentBuilder()
+ .withNewMetadata()
+ .withName(deploymentName)
+ .withNamespace(namespace)
+ .endMetadata()
+ .build();
+ DeploymentList deploymentList =
+ new DeploymentListBuilder()
+ .withMetadata(new ListMeta())
+ .withItems(deployment)
+ .build();
+
+ // GET always returns the deployment; watch never emits DELETED within
the test window
+ mockServer
+ .expect()
+ .get()
+ .withPath(getUrl)
+ .andReturn(HttpURLConnection.HTTP_OK, deploymentList)
+ .always();
+ mockServer
+ .expect()
+ .get()
+ .withPath(watchUrl)
+ .andUpgradeToWebSocket()
+ .open()
+ .waitFor(5000)
+ .andEmit(new WatchEvent(deployment, "DELETED"))
+ .done()
+ .always();
+
+ Assertions.assertThrows(
+ KubernetesClientTimeoutException.class,
+ () ->
+ AbstractFlinkService.deleteBlocking(
+ "Test",
+ () ->
+ client.apps()
+ .deployments()
+ .inNamespace(namespace)
+ .withName(deploymentName),
+ Duration.ofMillis(100)));
}
@ParameterizedTest
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 666f24da..4026bc8e 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
@@ -49,9 +49,14 @@ import
org.apache.flink.runtime.rest.messages.job.JobResourceRequirementsHeaders
import org.apache.flink.util.concurrent.Executors;
import io.fabric8.kubernetes.api.model.DeletionPropagation;
+import io.fabric8.kubernetes.api.model.ListMeta;
import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.api.model.WatchEvent;
import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;
+import io.fabric8.kubernetes.api.model.apps.DeploymentList;
+import io.fabric8.kubernetes.api.model.apps.DeploymentListBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientTimeoutException;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.junit.jupiter.api.BeforeEach;
@@ -75,6 +80,7 @@ import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConf
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -195,6 +201,64 @@ public class NativeFlinkServiceTest {
.get());
}
+ @Test
+ public void testDeleteClusterInternalThrowsOnDeploymentDeletionTimeout() {
+ var timeout = Duration.ofMillis(200);
+ configuration.set(
+
KubernetesOperatorConfigOptions.OPERATOR_RESOURCE_CLEANUP_TIMEOUT, timeout);
+ var flinkService =
+ new NativeFlinkService(
+ client,
+ null,
+ executorService,
+
FlinkOperatorConfiguration.fromConfiguration(configuration),
+ eventRecorder);
+
+ var dep =
+ new DeploymentBuilder()
+ .withNewMetadata()
+ .withName(TestUtils.TEST_DEPLOYMENT_NAME)
+ .withNamespace(TestUtils.TEST_NAMESPACE)
+ .endMetadata()
+ .build();
+ DeploymentList deploymentList =
+ new DeploymentListBuilder().withMetadata(new
ListMeta()).withItems(dep).build();
+
+ String getUrl =
+ String.format(
+
"/apis/apps/v1/namespaces/%s/deployments?fieldSelector=metadata.name%%3D%s",
+ TestUtils.TEST_NAMESPACE,
TestUtils.TEST_DEPLOYMENT_NAME);
+ String watchUrl =
+ String.format(
+
"/apis/apps/v1/namespaces/%s/deployments?allowWatchBookmarks=true&fieldSelector=metadata.name%%3D%s&timeoutSeconds=600&watch=true",
+ TestUtils.TEST_NAMESPACE,
TestUtils.TEST_DEPLOYMENT_NAME);
+
+ // GET always returns the deployment; watch never emits DELETED within
timeout
+ mockServer.expect().get().withPath(getUrl).andReturn(200,
deploymentList).always();
+ mockServer
+ .expect()
+ .get()
+ .withPath(watchUrl)
+ .andUpgradeToWebSocket()
+ .open()
+ .waitFor(5000)
+ .andEmit(new WatchEvent(dep, "DELETED"))
+ .done()
+ .always();
+
+ var deployment = TestUtils.buildApplicationCluster();
+ ReconciliationUtils.updateStatusForDeployedSpec(deployment, new
Configuration());
+
+ assertThrows(
+ KubernetesClientTimeoutException.class,
+ () ->
+ flinkService.deleteClusterInternal(
+ TestUtils.TEST_NAMESPACE,
+ TestUtils.TEST_DEPLOYMENT_NAME,
+ configManager.getObserveConfig(deployment),
+ DeletionPropagation.BACKGROUND));
+ }
+
@ParameterizedTest
@MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
public void testDeleteOnSavepointBefore1_15(FlinkVersion flinkVersion)
throws Exception {
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 a871466b..e4e245fa 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
@@ -29,23 +29,34 @@ import
org.apache.flink.kubernetes.operator.api.spec.JobSpec;
import org.apache.flink.kubernetes.operator.api.spec.KubernetesDeploymentMode;
import org.apache.flink.kubernetes.operator.artifact.ArtifactManager;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.utils.StandaloneKubernetesUtils;
import org.apache.flink.util.concurrent.Executors;
+import io.fabric8.kubernetes.api.model.DeletionPropagation;
+import io.fabric8.kubernetes.api.model.ListMeta;
import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.WatchEvent;
import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;
+import io.fabric8.kubernetes.api.model.apps.DeploymentList;
+import io.fabric8.kubernetes.api.model.apps.DeploymentListBuilder;
import io.fabric8.kubernetes.api.model.apps.DeploymentSpec;
+import io.fabric8.kubernetes.client.KubernetesClientTimeoutException;
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.time.Duration;
import java.util.List;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_HEALTH_PROBE_PORT;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -59,10 +70,10 @@ public class StandaloneFlinkServiceTest {
StandaloneFlinkService flinkStandaloneService;
FlinkConfigManager configManager;
FlinkDeployment flinkDeployment;
+ private final Configuration configuration = new Configuration();
@BeforeEach
public void setup() {
- var configuration = new Configuration();
configuration.set(KubernetesConfigOptions.CLUSTER_ID,
TestUtils.TEST_DEPLOYMENT_NAME);
configuration.set(KubernetesConfigOptions.NAMESPACE,
TestUtils.TEST_NAMESPACE);
configuration.set(OPERATOR_HEALTH_PROBE_PORT, 80);
@@ -261,6 +272,122 @@ public class StandaloneFlinkServiceTest {
assertFalse(service.getRuntimeConfig().containsKey(OPERATOR_HEALTH_PROBE_PORT.key()));
}
+ @Test
+ public void testDeleteClusterInternalThrowsOnDeploymentDeletionTimeout() {
+ var timeout = Duration.ofMillis(200);
+ configuration.set(
+
KubernetesOperatorConfigOptions.OPERATOR_RESOURCE_CLEANUP_TIMEOUT, timeout);
+ var service =
+ new StandaloneFlinkService(
+ kubernetesClient,
+ new ArtifactManager(configManager),
+ Executors.newDirectExecutorService(),
+
FlinkOperatorConfiguration.fromConfiguration(configuration));
+
+ String jmDeploymentName =
+ StandaloneKubernetesUtils.getJobManagerDeploymentName(
+ TestUtils.TEST_DEPLOYMENT_NAME);
+
+ var dep =
+ new DeploymentBuilder()
+ .withNewMetadata()
+ .withName(jmDeploymentName)
+ .withNamespace(TestUtils.TEST_NAMESPACE)
+ .endMetadata()
+ .build();
+ DeploymentList deploymentList =
+ new DeploymentListBuilder().withMetadata(new
ListMeta()).withItems(dep).build();
+
+ String getUrl =
+ String.format(
+
"/apis/apps/v1/namespaces/%s/deployments?fieldSelector=metadata.name%%3D%s",
+ TestUtils.TEST_NAMESPACE, jmDeploymentName);
+ String watchUrl =
+ String.format(
+
"/apis/apps/v1/namespaces/%s/deployments?allowWatchBookmarks=true&fieldSelector=metadata.name%%3D%s&timeoutSeconds=600&watch=true",
+ TestUtils.TEST_NAMESPACE, jmDeploymentName);
+
+ // GET always returns the deployment; watch never emits DELETED within
timeout
+ mockServer.expect().get().withPath(getUrl).andReturn(200,
deploymentList).always();
+ mockServer
+ .expect()
+ .get()
+ .withPath(watchUrl)
+ .andUpgradeToWebSocket()
+ .open()
+ .waitFor(5000)
+ .andEmit(new WatchEvent(dep, "DELETED"))
+ .done()
+ .always();
+
+ assertThrows(
+ KubernetesClientTimeoutException.class,
+ () ->
+ service.deleteClusterInternal(
+ TestUtils.TEST_NAMESPACE,
+ TestUtils.TEST_DEPLOYMENT_NAME,
+ new Configuration(),
+ DeletionPropagation.BACKGROUND));
+ }
+
+ @Test
+ public void testDeleteClusterInternalThrowsOnTaskManagerDeletionTimeout() {
+ var timeout = Duration.ofMillis(200);
+ configuration.set(
+
KubernetesOperatorConfigOptions.OPERATOR_RESOURCE_CLEANUP_TIMEOUT, timeout);
+ var service =
+ new StandaloneFlinkService(
+ kubernetesClient,
+ new ArtifactManager(configManager),
+ Executors.newDirectExecutorService(),
+
FlinkOperatorConfiguration.fromConfiguration(configuration));
+
+ String tmDeploymentName =
+ StandaloneKubernetesUtils.getTaskManagerDeploymentName(
+ TestUtils.TEST_DEPLOYMENT_NAME);
+
+ var dep =
+ new DeploymentBuilder()
+ .withNewMetadata()
+ .withName(tmDeploymentName)
+ .withNamespace(TestUtils.TEST_NAMESPACE)
+ .endMetadata()
+ .build();
+ DeploymentList deploymentList =
+ new DeploymentListBuilder().withMetadata(new
ListMeta()).withItems(dep).build();
+
+ String getUrl =
+ String.format(
+
"/apis/apps/v1/namespaces/%s/deployments?fieldSelector=metadata.name%%3D%s",
+ TestUtils.TEST_NAMESPACE, tmDeploymentName);
+ String watchUrl =
+ String.format(
+
"/apis/apps/v1/namespaces/%s/deployments?allowWatchBookmarks=true&fieldSelector=metadata.name%%3D%s&timeoutSeconds=600&watch=true",
+ TestUtils.TEST_NAMESPACE, tmDeploymentName);
+
+ // GET always returns the deployment; watch never emits DELETED within
timeout
+ mockServer.expect().get().withPath(getUrl).andReturn(200,
deploymentList).always();
+ mockServer
+ .expect()
+ .get()
+ .withPath(watchUrl)
+ .andUpgradeToWebSocket()
+ .open()
+ .waitFor(5000)
+ .andEmit(new WatchEvent(dep, "DELETED"))
+ .done()
+ .always();
+
+ assertThrows(
+ KubernetesClientTimeoutException.class,
+ () ->
+ service.deleteClusterInternal(
+ TestUtils.TEST_NAMESPACE,
+ TestUtils.TEST_DEPLOYMENT_NAME,
+ new Configuration(),
+ DeletionPropagation.BACKGROUND));
+ }
+
class TestingStandaloneFlinkService extends StandaloneFlinkService {
Configuration runtimeConfig;
