You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Lucas Borges (Jira)" <ji...@apache.org> on 2026/06/23 10:56:00 UTC
[jira] [Updated] (FLINK-39953) AbstractFlinkService.deleteBlocking() swallows KubernetesClientTimeoutException allowing cluster upgrade to proceed on top of still-running cluster
[ https://issues.apache.org/jira/browse/FLINK-39953?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Lucas Borges updated FLINK-39953:
---------------------------------
Description:
When a {{FlinkDeployment}} upgrade is triggered, the operator calls {{deleteClusterDeployment()}} before deploying the new cluster.
Internally, {{deleteClusterInternal()}} calls {{deleteDeploymentBlocking()}} then calling {{deleteBlocking() }}which issues the Kubernetes DELETE and finally {{waitUntilCondition(Objects::isNull, timeout, MILLISECONDS)}} on the fabric8 watch.
If the old cluster does not disappear within the configured {{kubernetes.operator.cluster.shutdown-timeout}} the fabric8 client throws {{{}KubernetesClientTimeoutException{}}}. Because any {{KubernetesClientException}} is swallowed in a generic catch block, only a WARN is emitted and the exception is *silently discarded:* {{updateStatusAfterClusterDeletion()}} runs, marking the cluster as deleted in the status, and the operator immediately proceeds to call {{{}submitApplicationCluster(){}}}, deploying the new version on top of a still-running cluster.
h2. Code Path
{code:java}
ApplicationReconciler.deploy()
└─ deleteClusterDeployment() ← calls deleteClusterInternal() + updateStatusAfterClusterDeletion()
└─ deleteClusterInternal()
└─ deleteDeploymentBlocking()
└─ deleteBlocking()
└─ waitUntilCondition(Objects::isNull, timeout)
→ throws KubernetesClientTimeoutException
→ caught by KubernetesClientException handler ← BUG: exception swallowed
└─ submitApplicationCluster() ← called unconditionally, runs against still-alive cluster {code}
h2. Reproducing the Issue
Deploy any flinkDeployment: {{{}flink-test-job{}}}, preferably with a low value for {{kubernetes.operator.resource.cleanup.timeout}}
Then, add a blocking finalizer to the underlying {{Deployment}} like so:
{code:java}
kubectl patch deployment flink-test-job --type=json -p='[{"op":"add","path":"/metadata/finalizers","value":["test/block-deletion"]}]' {code}
Finally, increment the restartNonce counter:
{code:java}
kubectl patch flinkdeployment flink-test-job --type=merge -p '{"spec":{"job":{"restartNonce": 123}}}'{code}
What the upgrade gets stuck with the following apache operator logs saying that the flink job already exists, which is misleading.
{code:java}
>>> Event[Job] | Warning | ERROR | Could not create Kubernetes cluster "flink-marwan-onboarding". -> Failure executing: POST at: https://172.17.0.1:443/apis/apps/v1/namespaces/flink-data-eng-infra/deployments. Message: object is being deleted: deployments.apps "flink-test-job" already exists {code}
h2. Proposed Fix
Add an explicit {{catch (KubernetesClientTimeoutException)}} block *{*}before{*}* the generic {{KubernetesClientException }}handler in {{AbstractFlinkService.deleteBlocking()}} and re-throw it (see PR for the fix).
If {{deleteBlocking()}} times out waiting for the old cluster to disappear, the exception propagates out of {{{}deploy(){}}}, the reconciler catches it as a {{ReconciliationException}} and JOSDK schedules an exponential-backoff retry. On retry the operator attempts deletion again. The new cluster is not submitted until the old one is confirmed gone.
was:
When a `FlinkDeployment` upgrade is triggered, the operator calls `deleteClusterDeployment()` before deploying the new cluster. Internally, `deleteClusterInternal()` calls `deleteDeploymentBlocking()` → `deleteBlocking()`, which issues the Kubernetes DELETE and then calls `waitUntilCondition(Objects::isNull, timeout, MILLISECONDS)` on the fabric8 watch.
If the old cluster does not disappear within the configured `kubernetes.operator.cluster.shutdown-timeout`, the fabric8 client throws `KubernetesClientTimeoutException`. Because `KubernetesClientTimeoutException extends KubernetesClientException` and overrides `getCode()` to return `0` (no HTTP status code), it falls into the existing generic catch block:
```
} catch (KubernetesClientException kce) {
// We completely ignore not found errors and simply log others
if (kce.getCode() != HttpURLConnection.HTTP_NOT_FOUND) {
LOG.warn("Error while " + operation, kce);
}
}
```
The condition `0 != 404` is true, so only a WARN is emitted and the exception is **silently discarded**. `updateStatusAfterClusterDeletion()` runs, marking the cluster as deleted in the status, and the operator immediately proceeds to call `submitApplicationCluster()`, deploying the new version on top of a still-running cluster.
### Code Path
```
ApplicationReconciler.deploy()
└─ deleteClusterDeployment() ← calls deleteClusterInternal() + updateStatusAfterClusterDeletion()
└─ deleteClusterInternal()
└─ deleteDeploymentBlocking()
└─ deleteBlocking()
└─ waitUntilCondition(Objects::isNull, timeout)
→ throws KubernetesClientTimeoutException
→ caught by KubernetesClientException handler ← BUG: exception swallowed
└─ submitApplicationCluster() ← called unconditionally, runs against still-alive cluster
```
### Operator Log Evidence
```
WARN AbstractFlinkService - Error while deleting JobManager Deployment
io.fabric8.kubernetes.client.KubernetesClientTimeoutException: Timed out waiting for ...
INFO NativeFlinkService - Deploying application cluster ← should NOT appear after timeout
```
### Proposed Fix
Add an explicit `catch (KubernetesClientTimeoutException)` block **before** the generic `KubernetesClientException`
handler in `AbstractFlinkService.deleteBlocking()` and re-throw it:
```java
} catch (KubernetesClientTimeoutException e) {
throw e;
} catch (KubernetesClientException kce) {
// We completely ignore not found errors and simply log others
if (kce.getCode() != HttpURLConnection.HTTP_NOT_FOUND) {
LOG.warn("Error while " + operation, kce);
}
}
```
If `deleteBlocking()` times out waiting for the old cluster to disappear, the exception propagates out of `deploy()`, the reconciler catches it as a `ReconciliationException`, sets the status to `UPGRADING` (already persisted before `deploy()` is entered), and JOSDK schedules an exponential-backoff retry. On retry the operator attempts deletion again. The new cluster is not submitted until the old one is confirmed gone.
The one callsite where a timeout should be absorbed rather than propagated is `shutdownJobManagersBlocking()` in `NativeFlinkService`, which performs an optional scale-to-zero step before the real deletion. That site catches `KubernetesClientTimeoutException` locally, logs a warning, and proceeds.
> AbstractFlinkService.deleteBlocking() swallows KubernetesClientTimeoutException allowing cluster upgrade to proceed on top of still-running cluster
> ---------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-39953
> URL: https://issues.apache.org/jira/browse/FLINK-39953
> Project: Flink
> Issue Type: Bug
> Components: Kubernetes Operator
> Affects Versions: kubernetes-operator-1.15.0
> Environment: Operator 1.12.1
> Flink 1.20.3
> Reporter: Lucas Borges
> Priority: Major
> Labels: pull-request-available
>
> When a {{FlinkDeployment}} upgrade is triggered, the operator calls {{deleteClusterDeployment()}} before deploying the new cluster.
> Internally, {{deleteClusterInternal()}} calls {{deleteDeploymentBlocking()}} then calling {{deleteBlocking() }}which issues the Kubernetes DELETE and finally {{waitUntilCondition(Objects::isNull, timeout, MILLISECONDS)}} on the fabric8 watch.
> If the old cluster does not disappear within the configured {{kubernetes.operator.cluster.shutdown-timeout}} the fabric8 client throws {{{}KubernetesClientTimeoutException{}}}. Because any {{KubernetesClientException}} is swallowed in a generic catch block, only a WARN is emitted and the exception is *silently discarded:* {{updateStatusAfterClusterDeletion()}} runs, marking the cluster as deleted in the status, and the operator immediately proceeds to call {{{}submitApplicationCluster(){}}}, deploying the new version on top of a still-running cluster.
> h2. Code Path
> {code:java}
> ApplicationReconciler.deploy()
> └─ deleteClusterDeployment() ← calls deleteClusterInternal() + updateStatusAfterClusterDeletion()
> └─ deleteClusterInternal()
> └─ deleteDeploymentBlocking()
> └─ deleteBlocking()
> └─ waitUntilCondition(Objects::isNull, timeout)
> → throws KubernetesClientTimeoutException
> → caught by KubernetesClientException handler ← BUG: exception swallowed
> └─ submitApplicationCluster() ← called unconditionally, runs against still-alive cluster {code}
> h2. Reproducing the Issue
> Deploy any flinkDeployment: {{{}flink-test-job{}}}, preferably with a low value for {{kubernetes.operator.resource.cleanup.timeout}}
> Then, add a blocking finalizer to the underlying {{Deployment}} like so:
> {code:java}
> kubectl patch deployment flink-test-job --type=json -p='[{"op":"add","path":"/metadata/finalizers","value":["test/block-deletion"]}]' {code}
>
>
> Finally, increment the restartNonce counter:
> {code:java}
> kubectl patch flinkdeployment flink-test-job --type=merge -p '{"spec":{"job":{"restartNonce": 123}}}'{code}
> What the upgrade gets stuck with the following apache operator logs saying that the flink job already exists, which is misleading.
> {code:java}
> >>> Event[Job] | Warning | ERROR | Could not create Kubernetes cluster "flink-marwan-onboarding". -> Failure executing: POST at: https://172.17.0.1:443/apis/apps/v1/namespaces/flink-data-eng-infra/deployments. Message: object is being deleted: deployments.apps "flink-test-job" already exists {code}
> h2. Proposed Fix
> Add an explicit {{catch (KubernetesClientTimeoutException)}} block *{*}before{*}* the generic {{KubernetesClientException }}handler in {{AbstractFlinkService.deleteBlocking()}} and re-throw it (see PR for the fix).
>
> If {{deleteBlocking()}} times out waiting for the old cluster to disappear, the exception propagates out of {{{}deploy(){}}}, the reconciler catches it as a {{ReconciliationException}} and JOSDK schedules an exponential-backoff retry. On retry the operator attempts deletion again. The new cluster is not submitted until the old one is confirmed gone.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)