[
https://issues.apache.org/jira/browse/FLINK-41157?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41157:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Labels: pull-request-available
>
> 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:java}
> 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 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}
> 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)
