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 c134cc66f [FLINK-40954] Force delete pods stuck Terminating on cluster
resourcecleanup timeout (#1232)
c134cc66f is described below
commit c134cc66f71c18671ead271a4a6829cb7381652b
Author: Gyula Fora <[email protected]>
AuthorDate: Wed Nov 4 10:08:45 2026 -0800
[FLINK-40954] Force delete pods stuck Terminating on cluster
resourcecleanup timeout (#1232)
---
docs/content.zh/docs/operations/events.md | 171 +------------
docs/content/docs/operations/events.md | 171 +------------
.../kubernetes_operator_config_configuration.html | 6 +
.../generated/system_reconcile_section.html | 6 +
.../config/KubernetesOperatorConfigOptions.java | 13 +
.../operator/service/AbstractFlinkService.java | 72 ++++++
.../operator/service/NativeFlinkService.java | 9 +-
.../operator/service/StandaloneFlinkService.java | 15 +-
.../operator/service/NativeFlinkServiceTest.java | 266 +++++++++++++++++++++
.../service/StandaloneFlinkServiceTest.java | 103 ++++++++
10 files changed, 503 insertions(+), 329 deletions(-)
diff --git a/docs/content.zh/docs/operations/events.md
b/docs/content.zh/docs/operations/events.md
index 0bacd1457..ca755692e 100644
--- a/docs/content.zh/docs/operations/events.md
+++ b/docs/content.zh/docs/operations/events.md
@@ -63,18 +63,19 @@ These events are emitted by the core reconcile and observe
loops that manage Fli
| Reason | Type | Component | Description
|
|---------------|--------|----------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `Submit` | Normal | JobManagerDeployment | A `FlinkDeployment` or
`FlinkSessionJob` is being submitted to the cluster (`Starting deployment`).
Emitted once per deployment attempt.
|
-| `SpecChanged` | Normal | JobManagerDeployment | The reconciler detected a
difference between the desired and last reconciled spec. See [message
formats](#operator-message-formats). **Dedup:** by resource generation, so each
generation produces exactly one event. |
+| `SpecChanged` | Normal | JobManagerDeployment | The reconciler detected a
difference between the desired and last reconciled spec. **Dedup:** by resource
generation, so each generation produces exactly one event. |
| `Suspended` | Normal | JobManagerDeployment | The existing deployment is
being suspended before an upgrade or spec-driven redeploy (`Suspending existing
deployment.`).
|
| `Rollback` | Normal | JobManagerDeployment | A failed upgrade is being
rolled back to the previous stable spec (`Rolling back failed deployment.`).
Triggered automatically when [rollback]({{< ref
"docs/managing/job-management#rollbacks" >}}) is enabled. |
| `Scaling` | Normal | Job | In-place parallelism change
applied to a running job via the Flink rescale API, without restart (`In-place
scaling triggered`).
|
| `Cleanup` | Normal | Operator | The resource is being
deleted (`Cleaning up FlinkDeployment` or `Cleaning up FlinkSessionJob`). The
operator is tearing down associated Kubernetes objects and Flink state.
|
+| `ForceDeleted` | Warning | Operator | Cluster resource cleanup (suspend or
deletion) timed out waiting TM/JM pod deletion. The operator force deleted
those pods (grace period 0) and keeps waiting for the rest of the timeout. |
### Job Status
| Reason | Type | Component | Description
|
|-----------------------|---------|---------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `JobStatusChanged` | Normal | Job | The Flink job
transitioned to a new state (e.g. `RUNNING`, `FINISHED`, `FAILED`, `CANCELED`).
See [message formats](#operator-message-formats).
|
-| `JobException` | Warning | Job | The job
manager reported an exception (root cause or vertex-level). Annotations record
the task name, task manager id and exception timestamp. See [message
formats](#operator-message-formats). **Dedup:** by a hash of the formatted
message (`messageKey = jobmanager-exception-<hash>`). |
+| `JobStatusChanged` | Normal | Job | The Flink job
transitioned to a new state (e.g. `RUNNING`, `FINISHED`, `FAILED`, `CANCELED`).
|
+| `JobException` | Warning | Job | The job
manager reported an exception (root cause or vertex-level). Annotations record
the task name, task manager id and exception timestamp. **Dedup:** by a hash of
the formatted message (`messageKey = jobmanager-exception-<hash>`). |
| `Missing` | Warning | Job, JobManagerDeployment | The job (`Job
Not Found`) or the JobManager deployment (`Missing JobManager deployment`)
expected to exist on the cluster could not be found. May indicate a lost
JobManager pod or a failed submission.
|
| `RecoverDeployment` | Warning | Job | The operator
detected that the JobManager deployment was lost and is attempting to recover
it from the last reconciled state (`Recovering lost deployment`).
|
| `RestartUnhealthyJob` | Warning | Job | The cluster
health check failed (e.g. restart-count threshold exceeded) and the operator is
restarting the job. The message is the error returned by the [health check]({{<
ref "docs/operations/health" >}}), or `Restarting unhealthy job` as a fallback.
|
@@ -84,9 +85,9 @@ These events are emitted by the core reconcile and observe
loops that manage Fli
| Reason | Type | Component | Description
|
|---------------------|---------|--------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `SavepointError` | Warning | Operator, Snapshot | A triggered savepoint
failed. The message is a fixed phrase indicating periodic vs. nonce-triggered
failure; the Flink-returned root cause is recorded in `status.error`, not in
the event message. Emitted by the periodic savepoint trigger, the upgrade path,
and the `FlinkStateSnapshot` controller. See [message
formats](#operator-message-formats). |
-| `CheckpointError` | Warning | Operator, Snapshot | A triggered checkpoint
failed. Same shape as `SavepointError`: the message is the templated phrase,
the root cause lives in `status.error`. Emitted by the periodic checkpoint
trigger and the `FlinkStateSnapshot` controller. See [message
formats](#operator-message-formats).
|
-| `SnapshotAbandoned` | Warning | Snapshot | A `FlinkStateSnapshot`
was abandoned because its referenced `FlinkDeployment`/`FlinkSessionJob` is no
longer reconcilable (deleted or not running). See [message
formats](#operator-message-formats).
|
+| `SavepointError` | Warning | Operator, Snapshot | A triggered savepoint
failed. The message is a fixed phrase indicating periodic vs. nonce-triggered
failure; the Flink-returned root cause is recorded in `status.error`, not in
the event message. Emitted by the periodic savepoint trigger, the upgrade path,
and the `FlinkStateSnapshot` controller. |
+| `CheckpointError` | Warning | Operator, Snapshot | A triggered checkpoint
failed. Same shape as `SavepointError`: the message is the templated phrase,
the root cause lives in `status.error`. Emitted by the periodic checkpoint
trigger and the `FlinkStateSnapshot` controller.
|
+| `SnapshotAbandoned` | Warning | Snapshot | A `FlinkStateSnapshot`
was abandoned because its referenced `FlinkDeployment`/`FlinkSessionJob` is no
longer reconcilable (deleted or not running).
|
### Validation and Configuration
@@ -95,100 +96,7 @@ These events are emitted by the core reconcile and observe
loops that manage Fli
| `ValidationError` | Warning | Operator, Snapshot | The submitted
spec failed one of the registered validators (e.g. invalid `upgradeMode`,
conflicting fields, malformed `FlinkStateSnapshot`). The message is the
validator-specific error string. Reconciliation is skipped until the spec is
fixed. |
| `UnsupportedFlinkVersion` | Warning | Operator | The
`spec.flinkVersion` is not supported by this operator release (`Flink version
<version> is not supported by this operator version`). See [Compatibility
Guarantees]({{< ref "docs/deployment/compatibility" >}}).
|
| `IngressManagement` | Warning | Operator | The resource
defines `spec.ingress` but ingress management is disabled in the operator
(`kubernetes.operator.ingress.manage`). The ingress section is ignored
(`Ingress management is turned off but ingress set in spec`).
|
-| `CleanupFailed` | Warning | Operator, Snapshot | Resource cleanup
could not complete. Common causes: a session `FlinkDeployment` still has child
`FlinkSessionJob` or non-terminated jobs, a job cancellation timed out, or a
savepoint required during cleanup failed. See [message
formats](#operator-message-formats). |
-
-### Operator Message Formats
-
-For events whose message follows a structured template, the verbatim formats
are listed below.
-
-#### SpecChanged
-
-```
-<DiffType> change(s) detected (<diff>), starting reconciliation.
-```
-Where `<DiffType>` is one of `UPGRADE`, `SCALE`, `SAVEPOINT_REDEPLOY`.
-
-#### JobStatusChanged
-
-```
-Job status changed from <old> to <new>
-```
-On the first observation (no previous state), the operator instead emits:
-
-```
-Job status changed to <state>
-```
-
-#### JobException
-
-```
-<root exception class>: <message>
-<first N lines of the stack trace>
-... (<N> more lines)
-```
-The number of stack-trace lines included is controlled by the operator
configuration; the trailing `... (<N> more lines)` suffix is only present when
the trace was truncated.
-
-#### SavepointError
-
-One of (depending on the trigger type that initiated the savepoint):
-
-```
-Periodic savepoint failed
-```
-
-```
-Savepoint failed for savepointTriggerNonce: <nonce>
-```
-
-The root cause returned by Flink is stored in `status.error`, not in the event
message.
-
-#### CheckpointError
-
-One of (depending on the trigger type that initiated the checkpoint):
-
-```
-Periodic checkpoint failed
-```
-
-```
-Checkpoint failed for checkpointTriggerNonce: <nonce>
-```
-
-The root cause returned by Flink is stored in `status.error`, not in the event
message.
-
-#### SnapshotAbandoned
-
-One of (depending on why the snapshot's secondary resource is no longer
reconcilable):
-
-```
-Secondary resource <jobReference> for savepoint <snapshotName> was not found
-```
-
-```
-Secondary resource <jobReference> for savepoint <snapshotName> is not running
-```
-
-Note: the literal word `savepoint` is used in both templates even when the
snapshot is a checkpoint.
-
-#### CleanupFailed
-
-For a session `FlinkDeployment` whose child resources are still present:
-
-```
-The session jobs [<sessionJobName1>, <sessionJobName2>, ...] should be deleted
first
-```
-
-For a session cluster with non-terminated unmanaged jobs (when
`kubernetes.operator.session.block-on-unmanaged-jobs` is enabled):
-
-```
-The session cluster has non terminated jobs [<jobId1>, <jobId2>, ...] that
should be cancelled first
-```
-
-For a `FlinkStateSnapshot` whose cleanup throws:
-
-```
-<exception message>
-```
+| `CleanupFailed` | Warning | Operator, Snapshot | Resource cleanup
could not complete. Common causes: a session `FlinkDeployment` still has child
`FlinkSessionJob` or non-terminated jobs, a job cancellation timed out, or a
savepoint required during cleanup failed. |
## Autoscaler Events
@@ -196,65 +104,13 @@ These events are emitted by the [Autoscaler]({{< ref
"docs/managing/autoscaler"
| Reason | Type | Component | Description
[...]
|------------------------|---------|-----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| `ScalingReport` | Normal | Operator | The autoscaler produced a
parallelism recommendation. The message is built from a header indicating
whether scaling execution is enabled or disabled, followed by one summary entry
per vertex. See [message formats](#autoscaler-message-formats). **Dedup:**
interval + hash of the recommended parallelism (a new event is emitted only
when the recommendation changes).
[...]
+| `ScalingReport` | Normal | Operator | The autoscaler produced a
parallelism recommendation. The message is built from a header indicating
whether scaling execution is enabled or disabled, followed by one summary entry
per vertex. **Dedup:** interval + hash of the recommended parallelism (a new
event is emitted only when the recommendation changes).
[...]
| `IneffectiveScaling` | Normal | Operator | A previous scaling action did
not produce the expected throughput increase (actual increase below
[`job.autoscaler.scaling.effectiveness.threshold`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}})). When
[`job.autoscaler.scaling.effectiveness.detection.enabled`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}}) is `true` the
autoscaler blocks further scale-ups on that vertex. **Dedup:** interval + [...]
| `ScalingLimited` | Warning | Operator | The recommended parallelism
for a vertex could not be applied as-is because of key-group and partition
alignment constraints, the configured `parallelismUpperLimit`, or the source
partition count, and the autoscaler fell back to the closest aligned value.
Emitted only by the deprecated legacy alignment modes
(`scaling.key-group.partitions.adjust.mode`); the current [alignment modes]({{<
ref "docs/managing/autoscaler#parallelism-alignment" [...]
| `MemoryPressure` | Normal | Operator | The job is under memory
pressure. Emitted when GC pressure exceeds
[`job.autoscaler.memory.gc-pressure.threshold`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}}) or heap usage
exceeds [`job.autoscaler.memory.heap-usage.threshold`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}}). Scaling is
paused entirely while this condition holds. **Dedup:** interval, with
`messageKey` distinguishing `gcPressu [...]
| `ResourceQuotaReached` | Warning | Operator | The recommended scaling
action would push the cluster beyond the configured resource quota
([`job.autoscaler.quota.cpu`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}}),
[`job.autoscaler.quota.memory`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}})). The action is
rejected. **Dedup:** by interval only.
[...]
| `AutoscalerError` | Warning | Operator | The autoscaler failed with an
unexpected exception during a scaling evaluation. The message is the exception
text, or an abbreviated stack trace (max 2048 chars) when no message is
available. **Dedup:** none.
[...]
-### Autoscaler Message Formats
-
-#### ScalingReport
-
-The header depends on whether scaling execution is enabled, gated by config,
or gated by an excluded period. One of:
-
-```
-Scaling execution enabled, begin scaling vertices: <entry1>, <entry2>, ...
-```
-
-```
-Scaling execution disabled by config job.autoscaler.scaling.enabled:false,
recommended parallelism change: <entry1>, <entry2>, ...
-```
-
-```
-Scaling execution disabled by config
job.autoscaler.excluded.periods:<periods>, recommended parallelism change:
<entry1>, <entry2>, ...
-```
-
-The header is followed by one entry per vertex:
-
-```
-{ Vertex ID <id> | Parallelism <old> -> <new> | Processing capacity <old> ->
<new> | Target data rate <rate> }
-```
-
-#### IneffectiveScaling
-
-```
-Ineffective scaling detected for <vertex> (expected increase: <X>, actual
increase <Y>). Blocking of ineffective scaling decisions is <enabled|disabled>
-```
-
-#### ScalingLimited
-
-```
-Scaling limited detected for <vertex> (expected parallelism: <X>, actual
parallelism <Y>). Scaling limited due to numKeyGroupsOrPartitions :
<N>，upperBoundForAlignment(maxParallelism or parallelismUpperLimit): <U>,
parallelismLowerLimit: <L>.
-```
-
-#### MemoryPressure
-
-```
-GC Pressure <value> is above the allowed limit for scaling operations. Please
adjust the available memory manually.
-```
-
-```
-Heap Usage <value> is above the allowed limit for scaling operations. Please
adjust the available memory manually.
-```
-
-#### ResourceQuotaReached
-
-```
-Resource usage is above the allowed limit for scaling operations. Please
adjust the resource quota manually.
-```
-
## Autotuning Events
[Scaling Autotuning]({{< ref "docs/managing/autoscaler#scaling-autotuning"
>}}) is part of the autoscaler and uses TaskManager memory metrics to recommend
a tuned memory configuration. The event is emitted regardless of whether tuning
is enabled, which allows previewing recommendations before opting in via
`job.autoscaler.memory.tuning.enabled`.
@@ -263,15 +119,6 @@ Resource usage is above the allowed limit for scaling
operations. Please adjust
|--------------------------------|--------|-----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `Configuration recommendation` | Normal | Operator | Memory tuning produced
a new TaskManager memory configuration. The message lists the recommended
values for managed memory, network memory, JVM overhead and metaspace, and
indicates whether the recommendation is auto-applied. **Dedup:** `messageKey`
`MemoryTuning`, on the `job.autoscaler.scaling.event.interval` window. |
-### Autotuning Message Format
-
-#### Configuration recommendation
-
-```
-Memory tuning recommends the following configuration (automatic tuning is
<enabled|disabled>):
-<formatted_config>
-```
-
## Event Reporters
Autoscaler events are routed through an `AutoScalerEventHandler`. The chosen
implementation depends on how the autoscaler is deployed. When the autoscaler
runs inside the operator on Kubernetes, the Kubernetes handler is wired in
directly. When the standalone autoscaler runs outside Kubernetes, configuration
picks one of two built-in alternatives (`LOGGING` or `JDBC`). The interface is
internal, with no extension point for custom handlers. Operator-side events go
through the Kubernetes r [...]
diff --git a/docs/content/docs/operations/events.md
b/docs/content/docs/operations/events.md
index 0bacd1457..ca755692e 100644
--- a/docs/content/docs/operations/events.md
+++ b/docs/content/docs/operations/events.md
@@ -63,18 +63,19 @@ These events are emitted by the core reconcile and observe
loops that manage Fli
| Reason | Type | Component | Description
|
|---------------|--------|----------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `Submit` | Normal | JobManagerDeployment | A `FlinkDeployment` or
`FlinkSessionJob` is being submitted to the cluster (`Starting deployment`).
Emitted once per deployment attempt.
|
-| `SpecChanged` | Normal | JobManagerDeployment | The reconciler detected a
difference between the desired and last reconciled spec. See [message
formats](#operator-message-formats). **Dedup:** by resource generation, so each
generation produces exactly one event. |
+| `SpecChanged` | Normal | JobManagerDeployment | The reconciler detected a
difference between the desired and last reconciled spec. **Dedup:** by resource
generation, so each generation produces exactly one event. |
| `Suspended` | Normal | JobManagerDeployment | The existing deployment is
being suspended before an upgrade or spec-driven redeploy (`Suspending existing
deployment.`).
|
| `Rollback` | Normal | JobManagerDeployment | A failed upgrade is being
rolled back to the previous stable spec (`Rolling back failed deployment.`).
Triggered automatically when [rollback]({{< ref
"docs/managing/job-management#rollbacks" >}}) is enabled. |
| `Scaling` | Normal | Job | In-place parallelism change
applied to a running job via the Flink rescale API, without restart (`In-place
scaling triggered`).
|
| `Cleanup` | Normal | Operator | The resource is being
deleted (`Cleaning up FlinkDeployment` or `Cleaning up FlinkSessionJob`). The
operator is tearing down associated Kubernetes objects and Flink state.
|
+| `ForceDeleted` | Warning | Operator | Cluster resource cleanup (suspend or
deletion) timed out waiting TM/JM pod deletion. The operator force deleted
those pods (grace period 0) and keeps waiting for the rest of the timeout. |
### Job Status
| Reason | Type | Component | Description
|
|-----------------------|---------|---------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `JobStatusChanged` | Normal | Job | The Flink job
transitioned to a new state (e.g. `RUNNING`, `FINISHED`, `FAILED`, `CANCELED`).
See [message formats](#operator-message-formats).
|
-| `JobException` | Warning | Job | The job
manager reported an exception (root cause or vertex-level). Annotations record
the task name, task manager id and exception timestamp. See [message
formats](#operator-message-formats). **Dedup:** by a hash of the formatted
message (`messageKey = jobmanager-exception-<hash>`). |
+| `JobStatusChanged` | Normal | Job | The Flink job
transitioned to a new state (e.g. `RUNNING`, `FINISHED`, `FAILED`, `CANCELED`).
|
+| `JobException` | Warning | Job | The job
manager reported an exception (root cause or vertex-level). Annotations record
the task name, task manager id and exception timestamp. **Dedup:** by a hash of
the formatted message (`messageKey = jobmanager-exception-<hash>`). |
| `Missing` | Warning | Job, JobManagerDeployment | The job (`Job
Not Found`) or the JobManager deployment (`Missing JobManager deployment`)
expected to exist on the cluster could not be found. May indicate a lost
JobManager pod or a failed submission.
|
| `RecoverDeployment` | Warning | Job | The operator
detected that the JobManager deployment was lost and is attempting to recover
it from the last reconciled state (`Recovering lost deployment`).
|
| `RestartUnhealthyJob` | Warning | Job | The cluster
health check failed (e.g. restart-count threshold exceeded) and the operator is
restarting the job. The message is the error returned by the [health check]({{<
ref "docs/operations/health" >}}), or `Restarting unhealthy job` as a fallback.
|
@@ -84,9 +85,9 @@ These events are emitted by the core reconcile and observe
loops that manage Fli
| Reason | Type | Component | Description
|
|---------------------|---------|--------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `SavepointError` | Warning | Operator, Snapshot | A triggered savepoint
failed. The message is a fixed phrase indicating periodic vs. nonce-triggered
failure; the Flink-returned root cause is recorded in `status.error`, not in
the event message. Emitted by the periodic savepoint trigger, the upgrade path,
and the `FlinkStateSnapshot` controller. See [message
formats](#operator-message-formats). |
-| `CheckpointError` | Warning | Operator, Snapshot | A triggered checkpoint
failed. Same shape as `SavepointError`: the message is the templated phrase,
the root cause lives in `status.error`. Emitted by the periodic checkpoint
trigger and the `FlinkStateSnapshot` controller. See [message
formats](#operator-message-formats).
|
-| `SnapshotAbandoned` | Warning | Snapshot | A `FlinkStateSnapshot`
was abandoned because its referenced `FlinkDeployment`/`FlinkSessionJob` is no
longer reconcilable (deleted or not running). See [message
formats](#operator-message-formats).
|
+| `SavepointError` | Warning | Operator, Snapshot | A triggered savepoint
failed. The message is a fixed phrase indicating periodic vs. nonce-triggered
failure; the Flink-returned root cause is recorded in `status.error`, not in
the event message. Emitted by the periodic savepoint trigger, the upgrade path,
and the `FlinkStateSnapshot` controller. |
+| `CheckpointError` | Warning | Operator, Snapshot | A triggered checkpoint
failed. Same shape as `SavepointError`: the message is the templated phrase,
the root cause lives in `status.error`. Emitted by the periodic checkpoint
trigger and the `FlinkStateSnapshot` controller.
|
+| `SnapshotAbandoned` | Warning | Snapshot | A `FlinkStateSnapshot`
was abandoned because its referenced `FlinkDeployment`/`FlinkSessionJob` is no
longer reconcilable (deleted or not running).
|
### Validation and Configuration
@@ -95,100 +96,7 @@ These events are emitted by the core reconcile and observe
loops that manage Fli
| `ValidationError` | Warning | Operator, Snapshot | The submitted
spec failed one of the registered validators (e.g. invalid `upgradeMode`,
conflicting fields, malformed `FlinkStateSnapshot`). The message is the
validator-specific error string. Reconciliation is skipped until the spec is
fixed. |
| `UnsupportedFlinkVersion` | Warning | Operator | The
`spec.flinkVersion` is not supported by this operator release (`Flink version
<version> is not supported by this operator version`). See [Compatibility
Guarantees]({{< ref "docs/deployment/compatibility" >}}).
|
| `IngressManagement` | Warning | Operator | The resource
defines `spec.ingress` but ingress management is disabled in the operator
(`kubernetes.operator.ingress.manage`). The ingress section is ignored
(`Ingress management is turned off but ingress set in spec`).
|
-| `CleanupFailed` | Warning | Operator, Snapshot | Resource cleanup
could not complete. Common causes: a session `FlinkDeployment` still has child
`FlinkSessionJob` or non-terminated jobs, a job cancellation timed out, or a
savepoint required during cleanup failed. See [message
formats](#operator-message-formats). |
-
-### Operator Message Formats
-
-For events whose message follows a structured template, the verbatim formats
are listed below.
-
-#### SpecChanged
-
-```
-<DiffType> change(s) detected (<diff>), starting reconciliation.
-```
-Where `<DiffType>` is one of `UPGRADE`, `SCALE`, `SAVEPOINT_REDEPLOY`.
-
-#### JobStatusChanged
-
-```
-Job status changed from <old> to <new>
-```
-On the first observation (no previous state), the operator instead emits:
-
-```
-Job status changed to <state>
-```
-
-#### JobException
-
-```
-<root exception class>: <message>
-<first N lines of the stack trace>
-... (<N> more lines)
-```
-The number of stack-trace lines included is controlled by the operator
configuration; the trailing `... (<N> more lines)` suffix is only present when
the trace was truncated.
-
-#### SavepointError
-
-One of (depending on the trigger type that initiated the savepoint):
-
-```
-Periodic savepoint failed
-```
-
-```
-Savepoint failed for savepointTriggerNonce: <nonce>
-```
-
-The root cause returned by Flink is stored in `status.error`, not in the event
message.
-
-#### CheckpointError
-
-One of (depending on the trigger type that initiated the checkpoint):
-
-```
-Periodic checkpoint failed
-```
-
-```
-Checkpoint failed for checkpointTriggerNonce: <nonce>
-```
-
-The root cause returned by Flink is stored in `status.error`, not in the event
message.
-
-#### SnapshotAbandoned
-
-One of (depending on why the snapshot's secondary resource is no longer
reconcilable):
-
-```
-Secondary resource <jobReference> for savepoint <snapshotName> was not found
-```
-
-```
-Secondary resource <jobReference> for savepoint <snapshotName> is not running
-```
-
-Note: the literal word `savepoint` is used in both templates even when the
snapshot is a checkpoint.
-
-#### CleanupFailed
-
-For a session `FlinkDeployment` whose child resources are still present:
-
-```
-The session jobs [<sessionJobName1>, <sessionJobName2>, ...] should be deleted
first
-```
-
-For a session cluster with non-terminated unmanaged jobs (when
`kubernetes.operator.session.block-on-unmanaged-jobs` is enabled):
-
-```
-The session cluster has non terminated jobs [<jobId1>, <jobId2>, ...] that
should be cancelled first
-```
-
-For a `FlinkStateSnapshot` whose cleanup throws:
-
-```
-<exception message>
-```
+| `CleanupFailed` | Warning | Operator, Snapshot | Resource cleanup
could not complete. Common causes: a session `FlinkDeployment` still has child
`FlinkSessionJob` or non-terminated jobs, a job cancellation timed out, or a
savepoint required during cleanup failed. |
## Autoscaler Events
@@ -196,65 +104,13 @@ These events are emitted by the [Autoscaler]({{< ref
"docs/managing/autoscaler"
| Reason | Type | Component | Description
[...]
|------------------------|---------|-----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| `ScalingReport` | Normal | Operator | The autoscaler produced a
parallelism recommendation. The message is built from a header indicating
whether scaling execution is enabled or disabled, followed by one summary entry
per vertex. See [message formats](#autoscaler-message-formats). **Dedup:**
interval + hash of the recommended parallelism (a new event is emitted only
when the recommendation changes).
[...]
+| `ScalingReport` | Normal | Operator | The autoscaler produced a
parallelism recommendation. The message is built from a header indicating
whether scaling execution is enabled or disabled, followed by one summary entry
per vertex. **Dedup:** interval + hash of the recommended parallelism (a new
event is emitted only when the recommendation changes).
[...]
| `IneffectiveScaling` | Normal | Operator | A previous scaling action did
not produce the expected throughput increase (actual increase below
[`job.autoscaler.scaling.effectiveness.threshold`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}})). When
[`job.autoscaler.scaling.effectiveness.detection.enabled`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}}) is `true` the
autoscaler blocks further scale-ups on that vertex. **Dedup:** interval + [...]
| `ScalingLimited` | Warning | Operator | The recommended parallelism
for a vertex could not be applied as-is because of key-group and partition
alignment constraints, the configured `parallelismUpperLimit`, or the source
partition count, and the autoscaler fell back to the closest aligned value.
Emitted only by the deprecated legacy alignment modes
(`scaling.key-group.partitions.adjust.mode`); the current [alignment modes]({{<
ref "docs/managing/autoscaler#parallelism-alignment" [...]
| `MemoryPressure` | Normal | Operator | The job is under memory
pressure. Emitted when GC pressure exceeds
[`job.autoscaler.memory.gc-pressure.threshold`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}}) or heap usage
exceeds [`job.autoscaler.memory.heap-usage.threshold`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}}). Scaling is
paused entirely while this condition holds. **Dedup:** interval, with
`messageKey` distinguishing `gcPressu [...]
| `ResourceQuotaReached` | Warning | Operator | The recommended scaling
action would push the cluster beyond the configured resource quota
([`job.autoscaler.quota.cpu`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}}),
[`job.autoscaler.quota.memory`]({{< ref
"docs/deployment/configuration#autoscaler-configuration" >}})). The action is
rejected. **Dedup:** by interval only.
[...]
| `AutoscalerError` | Warning | Operator | The autoscaler failed with an
unexpected exception during a scaling evaluation. The message is the exception
text, or an abbreviated stack trace (max 2048 chars) when no message is
available. **Dedup:** none.
[...]
-### Autoscaler Message Formats
-
-#### ScalingReport
-
-The header depends on whether scaling execution is enabled, gated by config,
or gated by an excluded period. One of:
-
-```
-Scaling execution enabled, begin scaling vertices: <entry1>, <entry2>, ...
-```
-
-```
-Scaling execution disabled by config job.autoscaler.scaling.enabled:false,
recommended parallelism change: <entry1>, <entry2>, ...
-```
-
-```
-Scaling execution disabled by config
job.autoscaler.excluded.periods:<periods>, recommended parallelism change:
<entry1>, <entry2>, ...
-```
-
-The header is followed by one entry per vertex:
-
-```
-{ Vertex ID <id> | Parallelism <old> -> <new> | Processing capacity <old> ->
<new> | Target data rate <rate> }
-```
-
-#### IneffectiveScaling
-
-```
-Ineffective scaling detected for <vertex> (expected increase: <X>, actual
increase <Y>). Blocking of ineffective scaling decisions is <enabled|disabled>
-```
-
-#### ScalingLimited
-
-```
-Scaling limited detected for <vertex> (expected parallelism: <X>, actual
parallelism <Y>). Scaling limited due to numKeyGroupsOrPartitions :
<N>，upperBoundForAlignment(maxParallelism or parallelismUpperLimit): <U>,
parallelismLowerLimit: <L>.
-```
-
-#### MemoryPressure
-
-```
-GC Pressure <value> is above the allowed limit for scaling operations. Please
adjust the available memory manually.
-```
-
-```
-Heap Usage <value> is above the allowed limit for scaling operations. Please
adjust the available memory manually.
-```
-
-#### ResourceQuotaReached
-
-```
-Resource usage is above the allowed limit for scaling operations. Please
adjust the resource quota manually.
-```
-
## Autotuning Events
[Scaling Autotuning]({{< ref "docs/managing/autoscaler#scaling-autotuning"
>}}) is part of the autoscaler and uses TaskManager memory metrics to recommend
a tuned memory configuration. The event is emitted regardless of whether tuning
is enabled, which allows previewing recommendations before opting in via
`job.autoscaler.memory.tuning.enabled`.
@@ -263,15 +119,6 @@ Resource usage is above the allowed limit for scaling
operations. Please adjust
|--------------------------------|--------|-----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `Configuration recommendation` | Normal | Operator | Memory tuning produced
a new TaskManager memory configuration. The message lists the recommended
values for managed memory, network memory, JVM overhead and metaspace, and
indicates whether the recommendation is auto-applied. **Dedup:** `messageKey`
`MemoryTuning`, on the `job.autoscaler.scaling.event.interval` window. |
-### Autotuning Message Format
-
-#### Configuration recommendation
-
-```
-Memory tuning recommends the following configuration (automatic tuning is
<enabled|disabled>):
-<formatted_config>
-```
-
## Event Reporters
Autoscaler events are routed through an `AutoScalerEventHandler`. The chosen
implementation depends on how the autoscaler is deployed. When the autoscaler
runs inside the operator on Kubernetes, the Kubernetes handler is wired in
directly. When the standalone autoscaler runs outside Kubernetes, configuration
picks one of two built-in alternatives (`LOGGING` or `JDBC`). The interface is
internal, with no extension point for custom handlers. Operator-side events go
through the Kubernetes r [...]
diff --git
a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
index 502cedb2e..be49fe50e 100644
---
a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
+++
b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
@@ -362,6 +362,12 @@
<td>Integer</td>
<td>The maximum number of threads running the reconciliation loop.
Use -1 for infinite.</td>
</tr>
+ <tr>
+
<td><h5>kubernetes.operator.resource.cleanup.force-delete-stuck-pods.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to force delete (grace period 0) the
JobManager/TaskManager pods that are still Terminating past their deletion
deadline (e.g. stuck due to their node becoming unreachable) while waiting for
the cluster resources to be deleted. Stuck pods are force deleted once, after
half of resource.cleanup.timeout has elapsed. Pods whose termination grace
period is longer than that are left alone, increase resource.cleanup.timeout
for those.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.resource.cleanup.timeout</h5></td>
<td style="word-wrap: break-word;">5 min</td>
diff --git a/docs/layouts/shortcodes/generated/system_reconcile_section.html
b/docs/layouts/shortcodes/generated/system_reconcile_section.html
index 5633ace90..318a5abe8 100644
--- a/docs/layouts/shortcodes/generated/system_reconcile_section.html
+++ b/docs/layouts/shortcodes/generated/system_reconcile_section.html
@@ -56,6 +56,12 @@
<td>Duration</td>
<td>The interval for the controller to reschedule the reconcile
process.</td>
</tr>
+ <tr>
+
<td><h5>kubernetes.operator.resource.cleanup.force-delete-stuck-pods.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to force delete (grace period 0) the
JobManager/TaskManager pods that are still Terminating past their deletion
deadline (e.g. stuck due to their node becoming unreachable) while waiting for
the cluster resources to be deleted. Stuck pods are force deleted once, after
half of resource.cleanup.timeout has elapsed. Pods whose termination grace
period is longer than that are left alone, increase resource.cleanup.timeout
for those.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.resource.cleanup.timeout</h5></td>
<td style="word-wrap: break-word;">5 min</td>
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
index fb98eba90..8245dca46 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -178,6 +178,19 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"The timeout for the resource clean up to wait for
flink to shutdown cluster.");
+ @Documentation.Section(SECTION_SYSTEM_RECONCILE)
+ public static final ConfigOption<Boolean>
OPERATOR_RESOURCE_CLEANUP_FORCE_DELETE_STUCK_PODS =
+ operatorConfig("resource.cleanup.force-delete-stuck-pods.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Whether to force delete (grace period 0) the
JobManager/TaskManager pods that are "
+ + "still Terminating past their deletion
deadline (e.g. stuck due to their node "
+ + "becoming unreachable) while waiting for
the cluster resources to be deleted. "
+ + "Stuck pods are force deleted once,
after half of resource.cleanup.timeout has "
+ + "elapsed. Pods whose termination grace
period is longer than that are left "
+ + "alone, increase
resource.cleanup.timeout for those.");
+
@Documentation.Section(SECTION_SYSTEM_RECONCILE)
public static final ConfigOption<Duration> OPERATOR_JOB_SUBMISSION_TIMEOUT
=
operatorConfig("job.submission.timeout")
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 693d7b985..3928a1898 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
@@ -54,6 +54,7 @@ import
org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.EnvUtils;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventUtils;
import org.apache.flink.kubernetes.operator.utils.ExceptionUtils;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
@@ -114,6 +115,7 @@ import
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseSt
import com.google.common.collect.Iterables;
import io.fabric8.kubernetes.api.model.DeletionPropagation;
import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodList;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -140,6 +142,7 @@ import java.net.URL;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.time.Duration;
+import java.time.Instant;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -172,6 +175,7 @@ public abstract class AbstractFlinkService implements
FlinkService {
private static final Logger LOG =
LoggerFactory.getLogger(AbstractFlinkService.class);
private static final String EMPTY_JAR_FILENAME = "empty.jar";
+ private static final String FORCE_DELETED_EVENT_KEY = "ForceDeleted";
public static final String FIELD_NAME_TOTAL_CPU = "total-cpu";
public static final String FIELD_NAME_TOTAL_MEMORY = "total-memory";
public static final String FIELD_NAME_STATE_SIZE = "state-size";
@@ -1242,6 +1246,74 @@ public abstract class AbstractFlinkService implements
FlinkService {
return conf;
}
+ @VisibleForTesting
+ protected void forceDeleteStuckPods(Map<String, String> labels, Deployment
deployment) {
+ var now = Instant.now();
+ var stuckPods =
+ kubernetesClient
+ .pods()
+ .inNamespace(deployment.getMetadata().getNamespace())
+ .withLabels(labels)
+ .list()
+ .getItems()
+ .stream()
+ .filter(pod -> isPastDeletionDeadline(pod, now))
+ .collect(Collectors.toList());
+ stuckPods.forEach(pod ->
kubernetesClient.resource(pod).withGracePeriod(0).delete());
+
+ if (!stuckPods.isEmpty()) {
+ EventUtils.createOrUpdateEventWithInterval(
+ kubernetesClient,
+ deployment,
+ EventRecorder.Type.Warning,
+ FORCE_DELETED_EVENT_KEY,
+ String.format(
+ "Force deleted %d pod(s) that did not terminate in
time "
+ + "(likely their node became unreachable),
blocking cluster resource cleanup.",
+ stuckPods.size()),
+ EventRecorder.Component.Operator,
+ event -> {},
+ FORCE_DELETED_EVENT_KEY,
+ null);
+ }
+ }
+
+ // The deletion timestamp of a pod already includes its termination grace
period
+ private static boolean isPastDeletionDeadline(Pod pod, Instant now) {
+ var deletionTimestamp = pod.getMetadata().getDeletionTimestamp();
+ return deletionTimestamp != null &&
now.isAfter(Instant.parse(deletionTimestamp));
+ }
+
+ protected Duration deleteDeploymentBlockingWithForceDelete(
+ String name,
+ Resource<Deployment> deployment,
+ DeletionPropagation propagation,
+ Duration timeout,
+ Map<String, String> podLabels,
+ Configuration conf) {
+ if (!conf.get(
+ KubernetesOperatorConfigOptions
+ .OPERATOR_RESOURCE_CLEANUP_FORCE_DELETE_STUCK_PODS)) {
+ return deleteDeploymentBlocking(name, deployment, propagation,
timeout);
+ }
+
+ // Give the pods half of the timeout to terminate on their own, then
force delete the ones
+ // that are stuck and wait for the rest of the timeout
+ var firstWait = timeout.dividedBy(2);
+ var remaining = timeout.minus(firstWait);
+
+ try {
+ return remaining.plus(
+ deleteDeploymentBlocking(name, deployment, propagation,
firstWait));
+ } catch (KubernetesClientTimeoutException timeoutException) {
+ var current = deployment.get();
+ if (current != null) {
+ forceDeleteStuckPods(podLabels, current);
+ }
+ return deleteDeploymentBlocking(name, deployment, propagation,
remaining);
+ }
+ }
+
/**
* Generic blocking delete operation implementation for triggering and
waiting for removal of
* the selected resources. By returning the remaining timeout we allow
chaining multiple delete
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 65c2c7cdf..3ca858ba7 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
@@ -169,7 +169,14 @@ public class NativeFlinkService extends
AbstractFlinkService {
shutdownJobManagersBlocking(
jmDeployment, namespace, clusterId,
remainingTimeout);
}
- deleteDeploymentBlocking("JobManager", jmDeployment,
deletionPropagation, remainingTimeout);
+
+ deleteDeploymentBlockingWithForceDelete(
+ "JobManager",
+ jmDeployment,
+ deletionPropagation,
+ remainingTimeout,
+ KubernetesUtils.getCommonLabels(clusterId),
+ conf);
}
@Override
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java
index c1888d604..6fb1f8d83 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/StandaloneFlinkService.java
@@ -147,11 +147,13 @@ public class StandaloneFlinkService extends
AbstractFlinkService {
.inNamespace(namespace)
.withName(StandaloneKubernetesUtils.getJobManagerDeploymentName(clusterId));
var remainingTimeout =
- deleteDeploymentBlocking(
+ deleteDeploymentBlockingWithForceDelete(
"JobManager",
jmDeployment,
deletionPropagation,
- operatorConfig.getFlinkShutdownClusterTimeout());
+ operatorConfig.getFlinkShutdownClusterTimeout(),
+
StandaloneKubernetesUtils.getJobManagerSelectors(clusterId),
+ conf);
var tmDeployment =
kubernetesClient
@@ -160,8 +162,13 @@ public class StandaloneFlinkService extends
AbstractFlinkService {
.inNamespace(namespace)
.withName(
StandaloneKubernetesUtils.getTaskManagerDeploymentName(clusterId));
- deleteDeploymentBlocking(
- "TaskManager", tmDeployment, deletionPropagation,
remainingTimeout);
+ deleteDeploymentBlockingWithForceDelete(
+ "TaskManager",
+ tmDeployment,
+ deletionPropagation,
+ remainingTimeout,
+ StandaloneKubernetesUtils.getTaskManagerSelectors(clusterId),
+ conf);
}
@Override
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 4026bc8e3..8b9239b8b 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,11 @@ 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.Event;
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.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;
@@ -59,6 +61,7 @@ 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 io.fabric8.mockwebserver.http.RecordedRequest;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
@@ -67,16 +70,21 @@ import org.junit.jupiter.params.provider.MethodSource;
import java.time.Duration;
import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
+import java.util.stream.Collectors;
import static
org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_HEALTH_PROBE_PORT;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -259,6 +267,264 @@ public class NativeFlinkServiceTest {
DeletionPropagation.BACKGROUND));
}
+ @Test
+ public void testDeleteClusterInternalForceDeletesStuckPods() throws
Exception {
+ configuration.set(
+
KubernetesOperatorConfigOptions.OPERATOR_RESOURCE_CLEANUP_TIMEOUT,
+ Duration.ofSeconds(4));
+ var flinkService =
+ new NativeFlinkService(
+ client,
+ null,
+ executorService,
+
FlinkOperatorConfiguration.fromConfiguration(configuration),
+ eventRecorder);
+
+ var dep = testDeployment();
+ var emptyList =
+ new DeploymentListBuilder().withMetadata(new
ListMeta()).withItems().build();
+ var podForceDeleted = new AtomicBoolean(false);
+ mockServer
+ .expect()
+ .get()
+ .withPath(
+ String.format(
+ "/apis/apps/v1/namespaces/%s/deployments/%s",
+ TestUtils.TEST_NAMESPACE,
TestUtils.TEST_DEPLOYMENT_NAME))
+ .andReturn(200, dep)
+ .always();
+ // The Deployment only disappears after the stuck pod was force deleted
+ mockServer
+ .expect()
+ .get()
+ .withPath(deploymentGetUrl())
+ .andReply(200, req -> podForceDeleted.get() ? emptyList :
deploymentList(dep))
+ .always();
+ mockServer
+ .expect()
+ .get()
+ .withPath(
+ 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))
+ .andUpgradeToWebSocket()
+ .open()
+ .waitFor(5000)
+ .andEmit(new WatchEvent(dep, "DELETED"))
+ .done()
+ .always();
+ createTerminatingPod("tm-1");
+ mockServer
+ .expect()
+ .delete()
+ .withPath(
+ String.format("/api/v1/namespaces/%s/pods/tm-1",
TestUtils.TEST_NAMESPACE))
+ .andReply(
+ 200,
+ req -> {
+ podForceDeleted.set(true);
+ return new PodBuilder()
+ .withNewMetadata()
+ .withName("tm-1")
+ .endMetadata()
+ .build();
+ })
+ .always();
+
+ var start = Instant.now();
+ assertDoesNotThrow(() -> deleteCluster(flinkService));
+
+ // Pods are only force deleted in the second half of the budget, but
before it ends
+ assertTrue(Duration.between(start, Instant.now()).toMillis() >= 2000);
+ assertTrue(Duration.between(start, Instant.now()).toSeconds() < 4);
+ assertEquals(1, forceDeletedEvents().size());
+ assertEquals("Operator",
forceDeletedEvents().get(0).getSource().getComponent());
+ }
+
+ @Test
+ public void testDeleteClusterInternalDoesNotForceDeleteWhenDisabled()
throws Exception {
+ configuration.set(
+
KubernetesOperatorConfigOptions.OPERATOR_RESOURCE_CLEANUP_FORCE_DELETE_STUCK_PODS,
+ false);
+ var flinkService = newFlinkServiceWithShortCleanupTimeout();
+ var dep = testDeployment();
+ mockDeploymentWait(dep, deploymentList(dep), -1);
+ createTerminatingPod("tm-1");
+
+ assertThrows(KubernetesClientTimeoutException.class, () ->
deleteCluster(flinkService));
+ assertTrue(forceDeletedEvents().isEmpty());
+ }
+
+ @Test
+ public void testForceDeleteStuckPods() throws Exception {
+ var flinkService = newFlinkServiceWithShortCleanupTimeout();
+ // Two pods are stuck past their deletion deadline, one is still
within its grace period
+ createTerminatingPod("tm-1");
+ createTerminatingPod("tm-2");
+ createPodWithDeletionTimestamp("tm-3", Instant.now().plusSeconds(300));
+ drainRequests();
+
+ forceDeleteStuckPods(flinkService);
+
+ // Only the overdue pods are deleted, without grace period
+ var deletes =
+ drainRequests().stream()
+ .filter(r -> "DELETE".equals(r.getMethod()))
+ .filter(r -> r.getPath().contains("/pods/"))
+ .collect(Collectors.toList());
+ assertEquals(2, deletes.size());
+ assertTrue(deletes.stream().noneMatch(r ->
r.getPath().endsWith("/pods/tm-3")));
+ for (var delete : deletes) {
+
assertTrue(delete.getBody().readUtf8().contains("\"gracePeriodSeconds\":0"));
+ }
+
+ var events = forceDeletedEvents();
+ assertEquals(1, events.size());
+ assertEquals(1, events.get(0).getCount());
+ assertTrue(events.get(0).getMessage().contains("2 pod(s)"));
+
+ // Later force deletions update the same event instead of creating new
ones
+ createTerminatingPod("tm-4");
+ forceDeleteStuckPods(flinkService);
+ events = forceDeletedEvents();
+ assertEquals(1, events.size());
+ assertEquals(2, events.get(0).getCount());
+ assertTrue(events.get(0).getMessage().contains("3 pod(s)"));
+ }
+
+ @Test
+ public void testForceDeleteStuckPodsWithoutPods() {
+ var flinkService = newFlinkServiceWithShortCleanupTimeout();
+
+ forceDeleteStuckPods(flinkService);
+ assertTrue(forceDeletedEvents().isEmpty());
+ }
+
+ private List<RecordedRequest> drainRequests() throws InterruptedException {
+ var requests = new ArrayList<RecordedRequest>();
+ RecordedRequest request;
+ while ((request = mockServer.takeRequest(100, TimeUnit.MILLISECONDS))
!= null) {
+ requests.add(request);
+ }
+ return requests;
+ }
+
+ private NativeFlinkService newFlinkServiceWithShortCleanupTimeout() {
+ configuration.set(
+
KubernetesOperatorConfigOptions.OPERATOR_RESOURCE_CLEANUP_TIMEOUT,
+ Duration.ofMillis(200));
+ return new NativeFlinkService(
+ client,
+ null,
+ executorService,
+ FlinkOperatorConfiguration.fromConfiguration(configuration),
+ eventRecorder);
+ }
+
+ private void deleteCluster(NativeFlinkService flinkService) {
+ flinkService.deleteClusterInternal(
+ TestUtils.TEST_NAMESPACE,
+ TestUtils.TEST_DEPLOYMENT_NAME,
+ configuration,
+ DeletionPropagation.BACKGROUND);
+ }
+
+ private void forceDeleteStuckPods(NativeFlinkService flinkService) {
+ flinkService.forceDeleteStuckPods(
+
KubernetesUtils.getCommonLabels(TestUtils.TEST_DEPLOYMENT_NAME),
testDeployment());
+ }
+
+ private Deployment testDeployment() {
+ return new DeploymentBuilder()
+ .withNewMetadata()
+ .withName(TestUtils.TEST_DEPLOYMENT_NAME)
+ .withNamespace(TestUtils.TEST_NAMESPACE)
+ .withUid("test-uid")
+ .endMetadata()
+ .build();
+ }
+
+ /** The finalizer keeps the pod in Terminating state like a pod on an
unreachable node. */
+ private void createPodWithDeletionTimestamp(String name, Instant
deletionTimestamp) {
+ client.resource(
+ new PodBuilder()
+ .withNewMetadata()
+ .withName(name)
+ .withNamespace(TestUtils.TEST_NAMESPACE)
+ .withLabels(
+ KubernetesUtils.getCommonLabels(
+
TestUtils.TEST_DEPLOYMENT_NAME))
+
.withDeletionTimestamp(deletionTimestamp.toString())
+ .endMetadata()
+ .build())
+ .create();
+ }
+
+ private void createTerminatingPod(String name) {
+ var pod =
+ new PodBuilder()
+ .withNewMetadata()
+ .withName(name)
+ .withNamespace(TestUtils.TEST_NAMESPACE)
+
.withLabels(KubernetesUtils.getCommonLabels(TestUtils.TEST_DEPLOYMENT_NAME))
+ .withFinalizers("test.io/block")
+ .endMetadata()
+ .build();
+ client.resource(pod).create();
+ client.resource(pod).delete();
+ }
+
+ private DeploymentList deploymentList(Deployment dep) {
+ return new DeploymentListBuilder().withMetadata(new
ListMeta()).withItems(dep).build();
+ }
+
+ private List<Event> forceDeletedEvents() {
+ return
client.v1().events().inNamespace(TestUtils.TEST_NAMESPACE).list().getItems().stream()
+ .filter(event -> "ForceDeleted".equals(event.getReason()))
+ .collect(Collectors.toList());
+ }
+
+ private String deploymentGetUrl() {
+ return String.format(
+
"/apis/apps/v1/namespaces/%s/deployments?fieldSelector=metadata.name%%3D%s",
+ TestUtils.TEST_NAMESPACE, TestUtils.TEST_DEPLOYMENT_NAME);
+ }
+
+ /** Mocks a Deployment wait that never sees the DELETED event, so it times
out. */
+ private void mockDeploymentWait(Deployment dep, DeploymentList list, int
times) {
+ var 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);
+ mockServer
+ .expect()
+ .get()
+ .withPath(
+ String.format(
+ "/apis/apps/v1/namespaces/%s/deployments/%s",
+ TestUtils.TEST_NAMESPACE,
TestUtils.TEST_DEPLOYMENT_NAME))
+ .andReturn(200, dep)
+ .always();
+ var getExpectation =
mockServer.expect().get().withPath(deploymentGetUrl());
+ var watchExpectation =
+ mockServer
+ .expect()
+ .get()
+ .withPath(watchUrl)
+ .andUpgradeToWebSocket()
+ .open()
+ .waitFor(5000)
+ .andEmit(new WatchEvent(dep, "DELETED"))
+ .done();
+ if (times < 0) {
+ getExpectation.andReturn(200, list).always();
+ watchExpectation.always();
+ } else {
+ getExpectation.andReturn(200, list).times(times);
+ watchExpectation.times(times);
+ }
+ }
+
@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 a20ee83a7..c99e5f2d1 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
@@ -38,6 +38,7 @@ 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.PodBuilder;
import io.fabric8.kubernetes.api.model.WatchEvent;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.api.model.apps.DeploymentBuilder;
@@ -55,8 +56,10 @@ import org.junit.jupiter.params.provider.CsvSource;
import java.time.Duration;
import java.util.List;
+import java.util.stream.Collectors;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_HEALTH_PROBE_PORT;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -321,6 +324,106 @@ public class StandaloneFlinkServiceTest {
DeletionPropagation.BACKGROUND));
}
+ @Test
+ public void testDeleteClusterInternalForceDeletesStuckJmPodAndRetries()
throws Exception {
+ 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 nonEmptyList =
+ new DeploymentListBuilder().withMetadata(new
ListMeta()).withItems(dep).build();
+ DeploymentList emptyList =
+ new DeploymentListBuilder()
+ .withMetadata(new ListMeta())
+ .withItems(List.of())
+ .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);
+
+ // First wait times out, after the force delete the Deployment is
reported as gone
+ mockServer
+ .expect()
+ .get()
+ .withPath(
+ String.format(
+ "/apis/apps/v1/namespaces/%s/deployments/%s",
+ TestUtils.TEST_NAMESPACE, jmDeploymentName))
+ .andReturn(200, dep)
+ .always();
+ mockServer.expect().get().withPath(getUrl).andReturn(200,
nonEmptyList).once();
+ mockServer
+ .expect()
+ .get()
+ .withPath(watchUrl)
+ .andUpgradeToWebSocket()
+ .open()
+ .waitFor(5000)
+ .andEmit(new WatchEvent(dep, "DELETED"))
+ .done()
+ .once();
+ mockServer.expect().get().withPath(getUrl).andReturn(200,
emptyList).always();
+
+ var stuckPod =
+ new PodBuilder()
+ .withNewMetadata()
+ .withName("jm-1")
+ .withNamespace(TestUtils.TEST_NAMESPACE)
+ .withLabels(
+
StandaloneKubernetesUtils.getJobManagerSelectors(
+ TestUtils.TEST_DEPLOYMENT_NAME))
+ .withFinalizers("test.io/block")
+ .endMetadata()
+ .build();
+ kubernetesClient.resource(stuckPod).create();
+ kubernetesClient.resource(stuckPod).delete();
+
+ assertDoesNotThrow(
+ () ->
+ service.deleteClusterInternal(
+ TestUtils.TEST_NAMESPACE,
+ TestUtils.TEST_DEPLOYMENT_NAME,
+ new Configuration(),
+ DeletionPropagation.BACKGROUND));
+
+ var events =
+ kubernetesClient
+ .v1()
+ .events()
+ .inNamespace(TestUtils.TEST_NAMESPACE)
+ .list()
+ .getItems()
+ .stream()
+ .filter(event ->
"ForceDeleted".equals(event.getReason()))
+ .collect(Collectors.toList());
+ assertEquals(1, events.size());
+ assertEquals(jmDeploymentName,
events.get(0).getInvolvedObject().getName());
+ assertEquals("Operator", events.get(0).getSource().getComponent());
+ }
+
class TestingStandaloneFlinkService extends StandaloneFlinkService {
Configuration runtimeConfig;
