[
https://issues.apache.org/jira/browse/FLINK-41157?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dennis-Mircea Ciupitu updated FLINK-41157:
------------------------------------------
Description:
h1. Summary
For a FlinkDeployment in application mode with {{job.upgradeMode: stateless}}
and Kubernetes HA enabled in {{{}spec.flinkConfiguration{}}}, the JobManager
deployment recovery path and the cluster health-check restart path require the
per-job HA ConfigMap ({{{}<cluster-id>--config-map{}}}) to exist. When it does
not, the operator fails the restart with {{RestoreFailed}} ("HA metadata is not
available to restore from last state") instead of redeploying the stateless job
from empty state. The eligibility checks of both paths explicitly treat a
stateless job as always restartable without HA metadata, so the gating and the
execution disagree.
Depending on the trigger, the deployment either stays {{FAILED}} and retries
the same failing restore on every reconcile, or ends up with no running cluster
at all while reporting lifecycle state {{STABLE}} and no error.
h1. Root cause
h2. Eligibility accepts stateless jobs without HA metadata
* {{{}AbstractFlinkResourceReconciler.shouldRecoverDeployment{}}}: a lost
JobManager deployment is recovered when the job is stateless or HA is enabled.
* {{{}ApplicationReconciler.shouldRestartJobBecauseUnhealthy{}}}: a stateless
job is always restarted, other upgrade modes only when
{{isHaMetadataAvailable}} is true.
h2. Execution requires HA metadata whenever HA is enabled
* {{ApplicationReconciler.reconcileOtherChanges}} calls {{resubmitJob(ctx,
HighAvailabilityMode.isHighAvailabilityModeActivated(observeConfig))}} - The
upgrade mode is not considered.
* {{AbstractJobReconciler.resubmitJob}} switches the spec to {{LAST_STATE}}
when {{requireHaMetadata}} is true and keeps the job id.
* {{AbstractFlinkService.submitApplicationCluster}} then calls
{{{}validateHaMetadataExists{}}}, which throws {{{}UpgradeFailureException("HA
metadata is not available to restore from last state. ...",
"RestoreFailed"){}}}.
h2. History
FLINK-31277 (operator 1.5.0) replaced the earlier call site {{requireHaMetadata
= upgradeMode != STATELESS}} with the HA-only condition and at the same time
widened {{shouldRecoverDeployment}} to "stateless or HA".
The combination: stateless + HA enabled + missing metadata has never been
covered.
h2. The availability check is existence-only
* {{FlinkUtils.isValidHaConfigMap}} accepts any ConfigMap whose name ends in
{{-config-map}} except {{{}-cluster-config-map{}}}, ignoring both content and
job id.
* {{FlinkUtils.checkpointExists}} is only used to decide whether the dummy
last-state savepoint path is set in {{{}deploy{}}}.
h2. When the per-job ConfigMap is missing
Flink creates it when the JobMaster builds its checkpoint store and deletes it
itself on global job cleanup. It is missing when a human or a cleanup script
removed the HA ConfigMaps, when the JobManager never reached job submission, or
after the job reached a terminal state (the FLINK-37392 flavour).
h1. Reproduction
The reproduction can be done on {{examples/basic-checkpoint-ha.yaml}} with a
single change: {{{}upgradeMode: stateless{}}}. On minikube the hostPath
additionally needs {{type: DirectoryOrCreate}} and a writable {{/tmp/flink}} on
the node.
Common steps:
* Apply the deployment and wait for {{jobStatus.state=RUNNING}} and
{{lifecycleState=STABLE}} + ConfigMaps {{{}-{}}}{{{}cluster-config-map{}}} and
{{-config-map}} exist.
* Delete the per-job ConfigMap: {{{}kubectl delete cm --config-map{}}}. The
running job is not affected (no checkpointing configured in the example).
h2. Trigger A: JobManager deployment recovery
Delete the deployment:
{code:bash}
kubectl delete deployment basic-checkpoint-ha-example
{code}
Observed on the next reconcile:
{code}
Observing JobManager deployment. Previous status: READY
Exception while getting job status
java.util.concurrent.TimeoutException
at java.base/java.util.concurrent.CompletableFuture.timedGet(Unknown
Source)
at java.base/java.util.concurrent.CompletableFuture.get(Unknown Source)
at
org.apache.flink.kubernetes.operator.service.AbstractFlinkService.getJobStatus(AbstractFlinkService.java:314)
at
org.apache.flink.kubernetes.operator.observer.JobStatusObserver.observe(JobStatusObserver.java:89)
at
org.apache.flink.kubernetes.operator.observer.deployment.ApplicationObserver.observeFlinkCluster(ApplicationObserver.java:49)
at
org.apache.flink.kubernetes.operator.observer.deployment.AbstractFlinkDeploymentObserver.observeInternal(AbstractFlinkDeploymentObserver.java:70)
at
org.apache.flink.kubernetes.operator.observer.AbstractFlinkResourceObserver.observe(AbstractFlinkResourceObserver.java:49)
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:145)
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:59)
at
io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:191)
at
io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:147)
at
org.apache.flink.kubernetes.operator.metrics.OperatorJosdkMetrics.timeControllerExecution(OperatorJosdkMetrics.java:82)
at
io.javaoperatorsdk.operator.processing.Controller.reconcile(Controller.java:146)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.reconcileExecution(ReconciliationDispatcher.java:164)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleReconcile(ReconciliationDispatcher.java:145)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleDispatch(ReconciliationDispatcher.java:111)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleExecution(ReconciliationDispatcher.java:76)
at
io.javaoperatorsdk.operator.processing.event.EventProcessor$ReconcilerExecutor.run(EventProcessor.java:578)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
Source)
at java.base/java.lang.Thread.run(Unknown Source)
Missing JobManager deployment
>>> Event[Job] | Warning | MISSING | Missing JobManager deployment
Config uses deprecated configuration key 'state.savepoints.dir' instead of
proper key 'execution.checkpointing.savepoint-dir'
>>> Event[Job] | Warning | RECOVERDEPLOYMENT | Recovering lost deployment
Resubmitting Flink job...
>>> Event[Job] | Info | SUBMIT | Starting deployment
Deploying application cluster requiring last-state from HA metadata
Job graph in ConfigMap stateless-ha-repro-cluster-config-map is deleted
>>> Event[Job] | Warning | RESTOREFAILED | HA metadata is not available
>>> to restore from last state.It is possible that the job has finished or
>>> terminally failed, or the configmaps have been deleted.
>>> Status[Job] | Error | FAILED | {...}
{code}
Status: {{{}jobStatus.state=RECONCILING{}}},
{{{}jobManagerDeploymentStatus=MISSING{}}}, {{{}lifecycleState=FAILED{}}},
{{error}} set to the {{{}UpgradeFailureException{}}}. The {{RecoverDeployment}}
/ {{Submit}} / {{RestoreFailed}} sequence repeated on every reconcile (event
count 5 after 3.5 minutes). No pods running.
h2. Trigger B: cluster health-check restart
Additional configuration on the deployment (within
{{{}spec.flinkConfiguration{}}}):
{code:java}
kubernetes.operator.cluster.health-check.enabled: "true"
kubernetes.operator.cluster.health-check.restarts.threshold: "1"
kubernetes.operator.cluster.health-check.restarts.window: 10 min
kubernetes.operator.cluster.health-check.checkpoint-progress.enabled: "false"
{code}
Delete the TaskManager pod twice, waiting for the new TaskManager to register
in between ({{{}numRestarts{}}} goes from 1 to 3). The following command was
used:
{code:java}
kubectl delete pod -l app=basic-checkpoint-ha-example,component=taskmanager
{code}
Observed within a single reconcile (operator log):
{code:java}
>>> Event[Job] | Info | JOBSTATUSCHANGED | Job status changed from
>>> RUNNING to RESTARTING
>>> Event[Job] | Warning | JOBEXCEPTION |
>>> org.apache.flink.util.FlinkExpectedException: The TaskExecutor is shutting
>>> down.
at
org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:523)
at
org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:255)
at
org.apache.flink.runtime.rpc.pekko.PekkoRpcActor$StartedState.lambda$terminate$0(PekkoRpcActor.java:583)
at
org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
... (22 more lines)
>>> Event[Job] | Warning | JOBEXCEPTION |
>>> org.apache.flink.util.FlinkExpectedException: The TaskExecutor is shutting
>>> down.
at
org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:523)
at
org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:255)
at
org.apache.flink.runtime.rpc.pekko.PekkoRpcActor$StartedState.lambda$terminate$0(PekkoRpcActor.java:583)
at
org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
... (22 more lines)
...
Restart count hit threshold: 1
>>> Event[Job] | Warning | RESTARTUNHEALTHYJOB | Restart count hit
>>> threshold: 1
Deleting cluster with Foreground propagation
Scaling JobManager Deployment to zero with 60 seconds timeout...
Completed Scaling JobManager Deployment to zero
Deleting JobManager Deployment with 298 seconds timeout...
Completed Deleting JobManager Deployment
Keeping HA metadata for last-state restore
Resubmitting Flink job...
>>> Event[Job] | Info | SUBMIT | Starting deployment
Deploying application cluster requiring last-state from HA metadata
Job graph in ConfigMap basic-checkpoint-ha-example-cluster-config-map is deleted
>>> Event[Job] | Warning | RESTOREFAILED | HA metadata is not available
>>> to restore from last state.It is possible that the job has finished or
>>> terminally failed, or the configmaps have been deleted.
>>> Status[Job] | Error | FAILED | {...}
{code}
End state: {{{}jobStatus.state=FINISHED{}}},
{{{}jobManagerDeploymentStatus=MISSING{}}}, {{{}lifecycleState=STABLE{}}},
{{error}} empty, no pods, {{RestoreFailed}} event count still 1 after 6 minutes.
In both triggers the attempted submit also removed the {{executionPlan-*}}
entry from {{-cluster-config-map}} (5 keys to 0) before the validation threw.
h1. Impact
The impact is different per each trigger:
* Trigger A: a stateless job with HA enabled cannot recover from a lost
JobManager deployment without manual intervention, and the operator repeats a
failing deploy attempt on every reconcile.
* Trigger B: a stateless job with HA enabled and the health check turned on is
terminated by the operator and never restarted, with no error recorded and
lifecycle {{{}STABLE{}}}. Alerting on {{lifecycleState}} or {{status.error}}
does not catch it.
The documented semantics of {{stateless}} ("upgrades from empty state") are not
honoured by the recovery paths.
h1. Workaround
The workaroung for the described issue is represented by any spec change, for
example bumping {{{}restartNonce{}}}.
* In the trigger A state (job {{{}RECONCILING{}}}) this cancels in stateless
mode, deletes the HA ConfigMaps and redeploys with a new job id (verified,
{{RUNNING}} about 30 s after the patch).
* In the trigger B state (job {{{}FINISHED{}}}) the suspend step is a no-op,
the cluster is redeployed with a new job id and the HA ConfigMaps are left
behind.
was:
h1. Summary
For a FlinkDeployment in application mode with {{job.upgradeMode: stateless}}
and Kubernetes HA enabled in {{{}spec.flinkConfiguration{}}}, the JobManager
deployment recovery path and the cluster health-check restart path require the
per-job HA ConfigMap ({{{}<cluster-id>--config-map{}}}) to exist. When it does
not, the operator fails the restart with {{RestoreFailed}} ("HA metadata is not
available to restore from last state") instead of redeploying the stateless job
from empty state. The eligibility checks of both paths explicitly treat a
stateless job as always restartable without HA metadata, so the gating and the
execution disagree.
Depending on the trigger, the deployment either stays {{FAILED}} and retries
the same failing restore on every reconcile, or ends up with no running cluster
at all while reporting lifecycle state {{STABLE}} and no error.
h1. Root cause
h2. Eligibility accepts stateless jobs without HA metadata
* {{{}AbstractFlinkResourceReconciler.shouldRecoverDeployment{}}}: a lost
JobManager deployment is recovered when the job is stateless or HA is enabled.
* {{{}ApplicationReconciler.shouldRestartJobBecauseUnhealthy{}}}: a stateless
job is always restarted, other upgrade modes only when
{{isHaMetadataAvailable}} is true.
h2. Execution requires HA metadata whenever HA is enabled
* {{ApplicationReconciler.reconcileOtherChanges}} calls {{resubmitJob(ctx,
HighAvailabilityMode.isHighAvailabilityModeActivated(observeConfig))}} - The
upgrade mode is not considered.
* {{AbstractJobReconciler.resubmitJob}} switches the spec to {{LAST_STATE}}
when {{requireHaMetadata}} is true and keeps the job id.
* {{AbstractFlinkService.submitApplicationCluster}} then calls
{{{}validateHaMetadataExists{}}}, which throws {{{}UpgradeFailureException("HA
metadata is not available to restore from last state. ...",
"RestoreFailed"){}}}.
h2. History
FLINK-31277 (operator 1.5.0) replaced the earlier call site {{requireHaMetadata
= upgradeMode != STATELESS}} with the HA-only condition and at the same time
widened {{shouldRecoverDeployment}} to "stateless or HA".
The combination: stateless + HA enabled + missing metadata has never been
covered.
h2. The availability check is existence-only
* {{FlinkUtils.isValidHaConfigMap}} accepts any ConfigMap whose name ends in
{{-config-map}} except {{{}-cluster-config-map{}}}, ignoring both content and
job id.
* {{FlinkUtils.checkpointExists}} is only used to decide whether the dummy
last-state savepoint path is set in {{{}deploy{}}}.
h2. When the per-job ConfigMap is missing
Flink creates it when the JobMaster builds its checkpoint store and deletes it
itself on global job cleanup. It is missing when a human or a cleanup script
removed the HA ConfigMaps, when the JobManager never reached job submission, or
after the job reached a terminal state (the FLINK-37392 flavour).
h1. Reproduction
The reproduction can be done on {{examples/basic-checkpoint-ha.yaml}} with a
single change: {{{}upgradeMode: stateless{}}}. On minikube the hostPath
additionally needs {{type: DirectoryOrCreate}} and a writable {{/tmp/flink}} on
the node.
Common steps:
* Apply the deployment and wait for {{jobStatus.state=RUNNING}} and
{{lifecycleState=STABLE}} + ConfigMaps {{{}-{}}}{{{}cluster-config-map{}}} and
{{-config-map}} exist.
* Delete the per-job ConfigMap: {{{}kubectl delete cm --config-map{}}}. The
running job is not affected (no checkpointing configured in the example).
h2. Trigger A: JobManager deployment recovery
Delete the deployment: {{{}kubectl delete deployment
{{basic-checkpoint-ha-example{}}}}}
Observed on the next reconcile:
{code}
Observing JobManager deployment. Previous status: READY
Exception while getting job status
java.util.concurrent.TimeoutException
at java.base/java.util.concurrent.CompletableFuture.timedGet(Unknown
Source)
at java.base/java.util.concurrent.CompletableFuture.get(Unknown Source)
at
org.apache.flink.kubernetes.operator.service.AbstractFlinkService.getJobStatus(AbstractFlinkService.java:314)
at
org.apache.flink.kubernetes.operator.observer.JobStatusObserver.observe(JobStatusObserver.java:89)
at
org.apache.flink.kubernetes.operator.observer.deployment.ApplicationObserver.observeFlinkCluster(ApplicationObserver.java:49)
at
org.apache.flink.kubernetes.operator.observer.deployment.AbstractFlinkDeploymentObserver.observeInternal(AbstractFlinkDeploymentObserver.java:70)
at
org.apache.flink.kubernetes.operator.observer.AbstractFlinkResourceObserver.observe(AbstractFlinkResourceObserver.java:49)
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:145)
at
org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:59)
at
io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:191)
at
io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:147)
at
org.apache.flink.kubernetes.operator.metrics.OperatorJosdkMetrics.timeControllerExecution(OperatorJosdkMetrics.java:82)
at
io.javaoperatorsdk.operator.processing.Controller.reconcile(Controller.java:146)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.reconcileExecution(ReconciliationDispatcher.java:164)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleReconcile(ReconciliationDispatcher.java:145)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleDispatch(ReconciliationDispatcher.java:111)
at
io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleExecution(ReconciliationDispatcher.java:76)
at
io.javaoperatorsdk.operator.processing.event.EventProcessor$ReconcilerExecutor.run(EventProcessor.java:578)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown
Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
Source)
at java.base/java.lang.Thread.run(Unknown Source)
Missing JobManager deployment
>>> Event[Job] | Warning | MISSING | Missing JobManager deployment
Config uses deprecated configuration key 'state.savepoints.dir' instead of
proper key 'execution.checkpointing.savepoint-dir'
>>> Event[Job] | Warning | RECOVERDEPLOYMENT | Recovering lost deployment
Resubmitting Flink job...
>>> Event[Job] | Info | SUBMIT | Starting deployment
Deploying application cluster requiring last-state from HA metadata
Job graph in ConfigMap stateless-ha-repro-cluster-config-map is deleted
>>> Event[Job] | Warning | RESTOREFAILED | HA metadata is not available
>>> to restore from last state.It is possible that the job has finished or
>>> terminally failed, or the configmaps have been deleted.
>>> Status[Job] | Error | FAILED | {...}
{code}
Status: {{{}jobStatus.state=RECONCILING{}}},
{{{}jobManagerDeploymentStatus=MISSING{}}}, {{{}lifecycleState=FAILED{}}},
{{error}} set to the {{{}UpgradeFailureException{}}}. The {{RecoverDeployment}}
/ {{Submit}} / {{RestoreFailed}} sequence repeated on every reconcile (event
count 5 after 3.5 minutes). No pods running.
h2. Trigger B: cluster health-check restart
Additional configuration on the deployment (within
{{{}spec.flinkConfiguration{}}}):
{code:java}
kubernetes.operator.cluster.health-check.enabled: "true"
kubernetes.operator.cluster.health-check.restarts.threshold: "1"
kubernetes.operator.cluster.health-check.restarts.window: 10 min
kubernetes.operator.cluster.health-check.checkpoint-progress.enabled: "false"
{code}
Delete the TaskManager pod twice, waiting for the new TaskManager to register
in between ({{{}numRestarts{}}} goes from 1 to 3). The following command was
used:
{code:java}
kubectl delete pod -l app=basic-checkpoint-ha-example,component=taskmanager
{code}
Observed within a single reconcile (operator log):
{code:java}
>>> Event[Job] | Info | JOBSTATUSCHANGED | Job status changed from
>>> RUNNING to RESTARTING
>>> Event[Job] | Warning | JOBEXCEPTION |
>>> org.apache.flink.util.FlinkExpectedException: The TaskExecutor is shutting
>>> down.
at
org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:523)
at
org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:255)
at
org.apache.flink.runtime.rpc.pekko.PekkoRpcActor$StartedState.lambda$terminate$0(PekkoRpcActor.java:583)
at
org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
... (22 more lines)
>>> Event[Job] | Warning | JOBEXCEPTION |
>>> org.apache.flink.util.FlinkExpectedException: The TaskExecutor is shutting
>>> down.
at
org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:523)
at
org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:255)
at
org.apache.flink.runtime.rpc.pekko.PekkoRpcActor$StartedState.lambda$terminate$0(PekkoRpcActor.java:583)
at
org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
... (22 more lines)
...
Restart count hit threshold: 1
>>> Event[Job] | Warning | RESTARTUNHEALTHYJOB | Restart count hit
>>> threshold: 1
Deleting cluster with Foreground propagation
Scaling JobManager Deployment to zero with 60 seconds timeout...
Completed Scaling JobManager Deployment to zero
Deleting JobManager Deployment with 298 seconds timeout...
Completed Deleting JobManager Deployment
Keeping HA metadata for last-state restore
Resubmitting Flink job...
>>> Event[Job] | Info | SUBMIT | Starting deployment
Deploying application cluster requiring last-state from HA metadata
Job graph in ConfigMap basic-checkpoint-ha-example-cluster-config-map is deleted
>>> Event[Job] | Warning | RESTOREFAILED | HA metadata is not available
>>> to restore from last state.It is possible that the job has finished or
>>> terminally failed, or the configmaps have been deleted.
>>> Status[Job] | Error | FAILED | {...}
{code}
End state: {{{}jobStatus.state=FINISHED{}}},
{{{}jobManagerDeploymentStatus=MISSING{}}}, {{{}lifecycleState=STABLE{}}},
{{error}} empty, no pods, {{RestoreFailed}} event count still 1 after 6 minutes.
In both triggers the attempted submit also removed the {{executionPlan-*}}
entry from {{-cluster-config-map}} (5 keys to 0) before the validation threw.
h1. Impact
The impact is different per each trigger:
* Trigger A: a stateless job with HA enabled cannot recover from a lost
JobManager deployment without manual intervention, and the operator repeats a
failing deploy attempt on every reconcile.
* Trigger B: a stateless job with HA enabled and the health check turned on is
terminated by the operator and never restarted, with no error recorded and
lifecycle {{{}STABLE{}}}. Alerting on {{lifecycleState}} or {{status.error}}
does not catch it.
The documented semantics of {{stateless}} ("upgrades from empty state") are not
honoured by the recovery paths.
h1. Workaround
The workaroung for the described issue is represented by any spec change, for
example bumping {{{}restartNonce{}}}.
* In the trigger A state (job {{{}RECONCILING{}}}) this cancels in stateless
mode, deletes the HA ConfigMaps and redeploys with a new job id (verified,
{{RUNNING}} about 30 s after the patch).
* In the trigger B state (job {{{}FINISHED{}}}) the suspend step is a no-op,
the cluster is redeployed with a new job id and the HA ConfigMaps are left
behind.
> Stateless jobs with HA enabled cannot be recovered or restarted when HA
> metadata is missing
> -------------------------------------------------------------------------------------------
>
> Key: FLINK-41157
> URL: https://issues.apache.org/jira/browse/FLINK-41157
> Project: Flink
> Issue Type: Bug
> Components: Kubernetes Operator
> Reporter: Dennis-Mircea Ciupitu
> Priority: Major
>
> h1. Summary
> For a FlinkDeployment in application mode with {{job.upgradeMode: stateless}}
> and Kubernetes HA enabled in {{{}spec.flinkConfiguration{}}}, the JobManager
> deployment recovery path and the cluster health-check restart path require
> the per-job HA ConfigMap ({{{}<cluster-id>--config-map{}}}) to exist. When it
> does not, the operator fails the restart with {{RestoreFailed}} ("HA metadata
> is not available to restore from last state") instead of redeploying the
> stateless job from empty state. The eligibility checks of both paths
> explicitly treat a stateless job as always restartable without HA metadata,
> so the gating and the execution disagree.
> Depending on the trigger, the deployment either stays {{FAILED}} and retries
> the same failing restore on every reconcile, or ends up with no running
> cluster at all while reporting lifecycle state {{STABLE}} and no error.
> h1. Root cause
> h2. Eligibility accepts stateless jobs without HA metadata
> * {{{}AbstractFlinkResourceReconciler.shouldRecoverDeployment{}}}: a lost
> JobManager deployment is recovered when the job is stateless or HA is enabled.
> * {{{}ApplicationReconciler.shouldRestartJobBecauseUnhealthy{}}}: a
> stateless job is always restarted, other upgrade modes only when
> {{isHaMetadataAvailable}} is true.
> h2. Execution requires HA metadata whenever HA is enabled
> * {{ApplicationReconciler.reconcileOtherChanges}} calls {{resubmitJob(ctx,
> HighAvailabilityMode.isHighAvailabilityModeActivated(observeConfig))}} - The
> upgrade mode is not considered.
> * {{AbstractJobReconciler.resubmitJob}} switches the spec to {{LAST_STATE}}
> when {{requireHaMetadata}} is true and keeps the job id.
> * {{AbstractFlinkService.submitApplicationCluster}} then calls
> {{{}validateHaMetadataExists{}}}, which throws
> {{{}UpgradeFailureException("HA metadata is not available to restore from
> last state. ...", "RestoreFailed"){}}}.
> h2. History
> FLINK-31277 (operator 1.5.0) replaced the earlier call site
> {{requireHaMetadata = upgradeMode != STATELESS}} with the HA-only condition
> and at the same time widened {{shouldRecoverDeployment}} to "stateless or HA".
> The combination: stateless + HA enabled + missing metadata has never been
> covered.
> h2. The availability check is existence-only
> * {{FlinkUtils.isValidHaConfigMap}} accepts any ConfigMap whose name ends in
> {{-config-map}} except {{{}-cluster-config-map{}}}, ignoring both content and
> job id.
> * {{FlinkUtils.checkpointExists}} is only used to decide whether the dummy
> last-state savepoint path is set in {{{}deploy{}}}.
> h2. When the per-job ConfigMap is missing
> Flink creates it when the JobMaster builds its checkpoint store and deletes
> it itself on global job cleanup. It is missing when a human or a cleanup
> script removed the HA ConfigMaps, when the JobManager never reached job
> submission, or after the job reached a terminal state (the FLINK-37392
> flavour).
> h1. Reproduction
> The reproduction can be done on {{examples/basic-checkpoint-ha.yaml}} with a
> single change: {{{}upgradeMode: stateless{}}}. On minikube the hostPath
> additionally needs {{type: DirectoryOrCreate}} and a writable {{/tmp/flink}}
> on the node.
> Common steps:
> * Apply the deployment and wait for {{jobStatus.state=RUNNING}} and
> {{lifecycleState=STABLE}} + ConfigMaps {{{}-{}}}{{{}cluster-config-map{}}}
> and {{-config-map}} exist.
> * Delete the per-job ConfigMap: {{{}kubectl delete cm --config-map{}}}. The
> running job is not affected (no checkpointing configured in the example).
> h2. Trigger A: JobManager deployment recovery
> Delete the deployment:
> {code:bash}
> kubectl delete deployment basic-checkpoint-ha-example
> {code}
> Observed on the next reconcile:
> {code}
> Observing JobManager deployment. Previous status: READY
> Exception while getting job status
> java.util.concurrent.TimeoutException
> at java.base/java.util.concurrent.CompletableFuture.timedGet(Unknown
> Source)
> at java.base/java.util.concurrent.CompletableFuture.get(Unknown
> Source)
> at
> org.apache.flink.kubernetes.operator.service.AbstractFlinkService.getJobStatus(AbstractFlinkService.java:314)
> at
> org.apache.flink.kubernetes.operator.observer.JobStatusObserver.observe(JobStatusObserver.java:89)
> at
> org.apache.flink.kubernetes.operator.observer.deployment.ApplicationObserver.observeFlinkCluster(ApplicationObserver.java:49)
> at
> org.apache.flink.kubernetes.operator.observer.deployment.AbstractFlinkDeploymentObserver.observeInternal(AbstractFlinkDeploymentObserver.java:70)
> at
> org.apache.flink.kubernetes.operator.observer.AbstractFlinkResourceObserver.observe(AbstractFlinkResourceObserver.java:49)
> at
> org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:145)
> at
> org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController.reconcile(FlinkDeploymentController.java:59)
> at
> io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:191)
> at
> io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:147)
> at
> org.apache.flink.kubernetes.operator.metrics.OperatorJosdkMetrics.timeControllerExecution(OperatorJosdkMetrics.java:82)
> at
> io.javaoperatorsdk.operator.processing.Controller.reconcile(Controller.java:146)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.reconcileExecution(ReconciliationDispatcher.java:164)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleReconcile(ReconciliationDispatcher.java:145)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleDispatch(ReconciliationDispatcher.java:111)
> at
> io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleExecution(ReconciliationDispatcher.java:76)
> at
> io.javaoperatorsdk.operator.processing.event.EventProcessor$ReconcilerExecutor.run(EventProcessor.java:578)
> at
> java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
> at
> java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
> at java.base/java.lang.Thread.run(Unknown Source)
> Missing JobManager deployment
> >>> Event[Job] | Warning | MISSING | Missing JobManager
> >>> deployment
> Config uses deprecated configuration key 'state.savepoints.dir' instead of
> proper key 'execution.checkpointing.savepoint-dir'
> >>> Event[Job] | Warning | RECOVERDEPLOYMENT | Recovering lost
> >>> deployment
> Resubmitting Flink job...
> >>> Event[Job] | Info | SUBMIT | Starting deployment
> Deploying application cluster requiring last-state from HA metadata
> Job graph in ConfigMap stateless-ha-repro-cluster-config-map is deleted
> >>> Event[Job] | Warning | RESTOREFAILED | HA metadata is not
> >>> available to restore from last state.It is possible that the job has
> >>> finished or terminally failed, or the configmaps have been deleted.
> >>> Status[Job] | Error | FAILED | {...}
> {code}
> Status: {{{}jobStatus.state=RECONCILING{}}},
> {{{}jobManagerDeploymentStatus=MISSING{}}}, {{{}lifecycleState=FAILED{}}},
> {{error}} set to the {{{}UpgradeFailureException{}}}. The
> {{RecoverDeployment}} / {{Submit}} / {{RestoreFailed}} sequence repeated on
> every reconcile (event count 5 after 3.5 minutes). No pods running.
> h2. Trigger B: cluster health-check restart
> Additional configuration on the deployment (within
> {{{}spec.flinkConfiguration{}}}):
> {code:java}
> kubernetes.operator.cluster.health-check.enabled: "true"
> kubernetes.operator.cluster.health-check.restarts.threshold: "1"
> kubernetes.operator.cluster.health-check.restarts.window: 10 min
> kubernetes.operator.cluster.health-check.checkpoint-progress.enabled: "false"
> {code}
> Delete the TaskManager pod twice, waiting for the new TaskManager to register
> in between ({{{}numRestarts{}}} goes from 1 to 3). The following command was
> used:
> {code:java}
> kubectl delete pod -l app=basic-checkpoint-ha-example,component=taskmanager
> {code}
> Observed within a single reconcile (operator log):
> {code:java}
> >>> Event[Job] | Info | JOBSTATUSCHANGED | Job status changed from
> >>> RUNNING to RESTARTING
> >>> Event[Job] | Warning | JOBEXCEPTION |
> >>> org.apache.flink.util.FlinkExpectedException: The TaskExecutor is
> >>> shutting down.
> at
> org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:523)
> at
> org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:255)
> at
> org.apache.flink.runtime.rpc.pekko.PekkoRpcActor$StartedState.lambda$terminate$0(PekkoRpcActor.java:583)
> at
> org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
> ... (22 more lines)
> >>> Event[Job] | Warning | JOBEXCEPTION |
> >>> org.apache.flink.util.FlinkExpectedException: The TaskExecutor is
> >>> shutting down.
> at
> org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:523)
> at
> org.apache.flink.runtime.rpc.RpcEndpoint.internalCallOnStop(RpcEndpoint.java:255)
> at
> org.apache.flink.runtime.rpc.pekko.PekkoRpcActor$StartedState.lambda$terminate$0(PekkoRpcActor.java:583)
> at
> org.apache.flink.runtime.concurrent.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
> ... (22 more lines)
> ...
> Restart count hit threshold: 1
> >>> Event[Job] | Warning | RESTARTUNHEALTHYJOB | Restart count hit
> >>> threshold: 1
> Deleting cluster with Foreground propagation
> Scaling JobManager Deployment to zero with 60 seconds timeout...
> Completed Scaling JobManager Deployment to zero
> Deleting JobManager Deployment with 298 seconds timeout...
> Completed Deleting JobManager Deployment
> Keeping HA metadata for last-state restore
> Resubmitting Flink job...
> >>> Event[Job] | Info | SUBMIT | Starting deployment
> Deploying application cluster requiring last-state from HA metadata
> Job graph in ConfigMap basic-checkpoint-ha-example-cluster-config-map is
> deleted
> >>> Event[Job] | Warning | RESTOREFAILED | HA metadata is not
> >>> available to restore from last state.It is possible that the job has
> >>> finished or terminally failed, or the configmaps have been deleted.
> >>> Status[Job] | Error | FAILED | {...}
> {code}
> End state: {{{}jobStatus.state=FINISHED{}}},
> {{{}jobManagerDeploymentStatus=MISSING{}}}, {{{}lifecycleState=STABLE{}}},
> {{error}} empty, no pods, {{RestoreFailed}} event count still 1 after 6
> minutes.
> In both triggers the attempted submit also removed the {{executionPlan-*}}
> entry from {{-cluster-config-map}} (5 keys to 0) before the validation threw.
> h1. Impact
> The impact is different per each trigger:
> * Trigger A: a stateless job with HA enabled cannot recover from a lost
> JobManager deployment without manual intervention, and the operator repeats a
> failing deploy attempt on every reconcile.
> * Trigger B: a stateless job with HA enabled and the health check turned on
> is terminated by the operator and never restarted, with no error recorded and
> lifecycle {{{}STABLE{}}}. Alerting on {{lifecycleState}} or {{status.error}}
> does not catch it.
> The documented semantics of {{stateless}} ("upgrades from empty state") are
> not honoured by the recovery paths.
> h1. Workaround
> The workaroung for the described issue is represented by any spec change, for
> example bumping {{{}restartNonce{}}}.
> * In the trigger A state (job {{{}RECONCILING{}}}) this cancels in stateless
> mode, deletes the HA ConfigMaps and redeploys with a new job id (verified,
> {{RUNNING}} about 30 s after the patch).
> * In the trigger B state (job {{{}FINISHED{}}}) the suspend step is a no-op,
> the cluster is redeployed with a new job id and the HA ConfigMaps are left
> behind.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
