# How Confluent Manager for Apache Flink Recovers Flink SQL Statements When Confluent Manager for Apache Flink® (CMF) starts, its startup recovery brings each Flink SQL statement back to the state that you left it in, as described in [What CMF recovers](overview.md#cmf-dr-what-is-recovered). A statement that was running keeps running, or starts again if its Flink resource is gone, and a stopped or finished statement stays as it is. If a statement cannot start with the state that it needs, or would process its input again, recovery stops it instead and says why, so that you can decide what to do. An interactive `SELECT` statement whose Flink resource is gone does not start again, because its results are not persisted. This topic uses the following terms: Flink resource : The Kubernetes resource that runs the job of a statement: a `FlinkSessionJob` on a shared compute pool, or a `FlinkDeployment` on a dedicated one. Meant to be running : Describes a statement whose job should run: you have not stopped or deleted the statement, and it has not completed or failed. Its phase is `RUNNING`, `PENDING`, `RECONCILING`, `RESTARTING`, or `FAILING`, or you resumed it and its job has not started yet. Restore point : The savepoint or checkpoint that the job of a recreated statement starts from. CMF chooses it as described in [How CMF chooses the restore point](#cmf-dr-restore-point). Parked : Stopped by startup recovery instead of started, because the statement cannot start with the state that it needs, or would process its input again. The status trait `parkedByRecovery` and the `parked` outcome of the `cmf_metadata_sync_objects_total` metric use this term. See [Statements that recovery parks](#cmf-dr-parked). ## Check your statements after recovery When startup recovery has finished, check the phase and the `status.detail` of each statement, for example with the command in [Verify the recovery](recover.md#cmf-dr-verify): | What you see | What it means | What to do | |-------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `RUNNING` | The job of the statement runs. If recovery recreated the statement, its job started from
the restore point of the statement. | Nothing. To check which restore point a recreated statement started from, see
[Verify the recovery](recover.md#cmf-dr-verify). | | `PENDING`, with the detail `Recreating the Flink resource after CMF restart.` | Recovery recreated the Flink resource of the statement, and its job is starting. | Wait until it is `RUNNING`. If it stays `PENDING`, see [Troubleshoot Disaster Recovery in Confluent Manager for Apache Flink](troubleshoot.md#cmf-dr-troubleshoot). | | `PENDING`, with a detail that starts with `Not recovered after CMF restart` | Recovery could not recreate or reapply the Flink resource of the statement, for example
because its environment or compute pool was not recovered, or because its stored plan
cannot be read. The detail says why. | Fix the cause, and restart CMF, as described in [Restart CMF](recover.md#cmf-dr-restart). The next startup
recovery tries again. | | `STOPPED`, with the status trait `parkedByRecovery` | Recovery parked the statement: it cannot start with the state that it needs, or would
process its input again. | Look up its reason in [Statements that recovery parks](#cmf-dr-parked), and resume or recreate the statement as
described there. | | `FAILED` | The statement has ended, and cannot be updated or resumed. Recovery sets a statement to
`FAILED` when its Flink resource is gone and it has no usable plan, or is an interactive
`SELECT` statement, whose results were not persisted. The detail says why. | Create a new statement, or recreate this one as described in [Recreate a statement](#cmf-dr-recreate). | For other symptoms, and for problems whose cause or fix is less obvious, see [Troubleshoot Disaster Recovery in Confluent Manager for Apache Flink](troubleshoot.md#cmf-dr-troubleshoot). ## What startup recovery does with each statement Startup recovery does the following with each statement: | Statement | What startup recovery does | |---------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | Meant to be running, Flink resource exists, for example after a routine restart | Reapplies the resource, except in the cases listed after the table. The phase stays as it
is. | | Meant to be running, Flink resource is missing | Recreates the resource, and rebuilds the plan `ConfigMap` from the stored copy if it is
gone. The job starts from the restore point of the statement. The statement is
`PENDING`, with the detail `Recreating the Flink resource after CMF restart.`, until its
job runs. If recovery cannot recreate the resource, the statement is `PENDING`,
`STOPPED`, or `FAILED`, as described in [Check your statements after recovery](#cmf-dr-statements-check). | | Interactive `SELECT` statement, Flink resource is missing | Sets it to `FAILED`, because its results are not persisted. A `SELECT` statement whose
results were all fetched is set to `COMPLETED` instead, or to `STOPPED` if it was being
stopped. | | Stopped, or on a suspended shared compute pool | Leaves it as it is. If its Flink resource is gone, resuming the statement creates it, with
the restore point described in [How CMF chooses the restore point](#cmf-dr-restore-point). On a suspended compute pool,
resume the pool first. A statement whose stop was in progress when its Flink resource was
lost is set to `STOPPED`. | | Completed or failed | Leaves it as it is. | | Being deleted | Completes the deletion when its Flink resource is gone. | Startup recovery does not reapply the existing Flink resource of a statement that is meant to be running in cases such as the following: - The environment or the compute pool of the statement was not recovered. - Startup recovery did not update the session cluster of the compute pool, as described in [What CMF recovers](overview.md#cmf-dr-what-is-recovered). - The plan `ConfigMap` of the statement is gone or unusable, and no usable copy of the plan is stored. For details, see [Troubleshoot Disaster Recovery in Confluent Manager for Apache Flink](troubleshoot.md#cmf-dr-troubleshoot). - The statement is an interactive `SELECT` statement. ## How CMF chooses the restore point When startup recovery recreates the Flink resource of a statement, the job starts from the first of the following that applies: 1. A new `startFromSavepoint`: one that you set or changed, or set again by changing its `savepointRedeployNonce`, and that no job has started from yet. If it no longer resolves, startup recovery stops the statement instead of falling back to another restore point. 2. The latest completed savepoint of the statement, regardless of its source (`MANUAL`, `SCHEDULE`, or `UPGRADE`) and regardless of the `upgradeMode` of the statement, so even a `stateless` statement keeps the state that it saved. After a job has started from a `startFromSavepoint`, only the savepoints taken since then count as the latest savepoint of the statement, as described in [Start a job from a savepoint](../jobs/savepoints.md#cmf-savepoints-start). 3. The `startFromSavepoint` that a job last started from, if it still resolves. If it does not, CMF logs a warning and does not use it. 4. None of the above: the `upgradeMode` decides. A `stateless` statement starts without state. A `savepoint` or `last-state` statement is not started: startup recovery stops it. For the `upgradeMode` of a statement that does not set `spec.upgradeMode`, see [Manage Statements in Confluent Manager for Apache Flink](../jobs/sql-statements/manage-statements.md#flink-sql-statements-manage). #### NOTE When CMF restores the latest savepoint of a statement, it starts the job with `allowNonRestoredState` set to `false`, so a savepoint that does not match the plan of the statement fails the job instead of dropping state. A restore from a `startFromSavepoint` uses the `allowNonRestoredState` of the `startFromSavepoint`, which is `false` unless you set it. Startup recovery also stops some `stateless` and batch statements. For details, see [Statements that recovery parks](#cmf-dr-parked). ### Limit the age of restore points A savepoint can be old, especially a manual one, and restoring it processes all input again since it was taken, or skips input that your sources no longer retain. To skip restore points older than a limit, set the `cmf.recovery.maxRestorePointAge` Helm value to a duration, such as `7d`. It is not set by default. Log messages call it `cmf.recovery.max-restore-point-age`. ```yaml cmf: recovery: maxRestorePointAge: 7d ``` The limit works as follows: - It applies only to the restore points that CMF chooses by itself: the latest savepoint of the statement, and the `startFromSavepoint` that a job last started from. It never applies to a new `startFromSavepoint`, because you asked for it. - CMF skips a restore point that is older than the limit. If it skips them all, the statement has no restore point, and the `upgradeMode` decides. - A `startFromSavepoint` given as a path that is not one of the savepoints of the statement, such as a checkpoint directory, has no age that CMF knows. CMF counts it as recent enough, unless it skipped the latest savepoint of the statement for its age. ## Statements that recovery parks When a statement cannot start with the state that it needs, or would process its input again, startup recovery parks it. The statement is `STOPPED`, its `status.detail` says why and what to do, and its `status.traits.parkedByRecovery` holds one of the following reasons: | Reason | Why | How to resume | |-----------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `no-restore-point` | The `upgradeMode` of the statement is `savepoint` or `last-state`, and it has no
savepoint that CMF can restore: it has none, or each was taken before its job last
started from a `startFromSavepoint` that no longer resolves. | [From a savepoint](#cmf-dr-resume-savepoint), [from the latest checkpoint](#cmf-dr-resume-checkpoint), or [without state](#cmf-dr-resume-stateless). | | `restore-point-too-old` | The `upgradeMode` of the statement is `savepoint` or `last-state`, and each of its
restore points is older than `cmf.recovery.maxRestorePointAge`. | [From a savepoint](#cmf-dr-resume-savepoint), [from the latest checkpoint](#cmf-dr-resume-checkpoint), or [without state](#cmf-dr-resume-stateless). | | `start-from-savepoint-unresolved` | The savepoint that an update set in `startFromSavepoint` no longer resolves. | [From a savepoint](#cmf-dr-resume-savepoint) or [from the latest checkpoint](#cmf-dr-resume-checkpoint), with a `startFromSavepoint` that resolves, or without
`startFromSavepoint`, so that CMF chooses the restore point. | | `carry-over-replay` | The statement is `stateless` and has no restore point, and its Kafka sources start from
offsets that it inherited through `sql.tables.initial-offset-from`, so starting it
would process all input again since those offsets. | [From a savepoint](#cmf-dr-resume-savepoint), [from the latest checkpoint](#cmf-dr-resume-checkpoint), or [without state](#cmf-dr-resume-stateless). | | `batch-rerun` | The statement runs in batch mode, and its job already ran. Starting it again would run
the job from the start, and could write its output a second time. | It cannot be resumed. To run it again, create a new statement, or recreate this one as
described in [Recreate a statement](#cmf-dr-recreate). | For example, the status of a statement without a restore point includes the following fields: ```json { "status": { "phase": "STOPPED", "detail": "Not recovered after CMF restart: no savepoint to restore from. Resume with startFromSavepoint, or set upgradeMode: stateless to start without state.", "traits": { "parkedByRecovery": "no-restore-point" } } } ``` A restart of CMF does not start a parked statement: it stays stopped until you resume it. To find parked statements, use any of the following: - The log of the startup recovery that parked them lists them among the resources that did not recover, with a reason that starts with `stopped instead of deployed`. - Each later startup recovery lists them in the warning `Statements remain stopped because an earlier recovery could not start them with their state`, and counts them with the `parked` outcome of the `cmf_metadata_sync_objects_total` metric. - The command in [Verify the recovery](recover.md#cmf-dr-verify) lists their `parkedByRecovery` reasons. CMF clears `parkedByRecovery` in the following cases: - You resume the statement. - CMF sets the statement to `FAILED` because its compute pool was deleted. ## Resume a parked statement You resume a statement with a `PUT` request that sets `spec.stopped` to `false`, as described in [Manage Statements in Confluent Manager for Apache Flink](../jobs/sql-statements/manage-statements.md#flink-sql-statements-manage). To choose the state that it starts from, set `spec.startFromSavepoint` or `spec.upgradeMode` in the same request. Build the request from the statement as CMF returns it, without the fields that CMF sets: a request that includes `metadata.updateTimestamp` is rejected with HTTP 422. For example: ```bash curl -s http://cmf:8080/cmf/api/v1/environments/env-1/statements/stmt-1 \ | jq 'del(.status, .result, .metadata.uid, .metadata.creationTimestamp, .metadata.updateTimestamp) | .spec.stopped = false' \ > stmt-1.json ``` Then set `spec.startFromSavepoint` or `spec.upgradeMode` in `stmt-1.json`, as described in the following sections, and send it: ```bash curl -X PUT -H "Content-Type: application/json" \ http://cmf:8080/cmf/api/v1/environments/env-1/statements/stmt-1 \ -d @stmt-1.json ``` When CMF cannot start the statement as you asked, it rejects the request and the statement stays stopped. For example, a resume without a restore point is rejected with HTTP 422 and a message such as `Its Flink resource no longer exists, and there is no savepoint to restore from.` ### Resume from a savepoint To restore a savepoint of the statement, name it in `startFromSavepoint`. If `startFromSavepoint` already names that savepoint, also change `savepointRedeployNonce` to a new value. To list the statement’s savepoints, use `GET /cmf/api/v1/environments/{envName}/statements/{stmtName}/savepoints`. The following example shows only the fields of the request that resume the statement from a savepoint: ```json { "spec": { "stopped": false, "startFromSavepoint": { "savepointName": "orders-sp-41" } } } ``` ### Resume from the latest checkpoint Flink stores the checkpoints of a job under `//`, unless `state.checkpoints.create-subdir` is `false`. A job whose cluster was lost was not canceled, so its latest completed checkpoints remain in checkpoint storage. To resume a statement that recovery stopped from the latest checkpoint of its last job: 1. Read the ID of the statement’s last job from `status.traits.jobId` of the statement. CMF keeps it while the statement is stopped, and a statement that recovery stopped has not started a new job, so this is the job that was lost. 2. In your checkpoint storage, find the checkpoint directory `chk-` with the highest `` under `//` that contains a `_metadata` file. A directory without one holds a checkpoint that did not complete. For example, with Amazon S3: ```bash aws s3 ls s3://my-bucket/checkpoints// aws s3 ls s3://my-bucket/checkpoints//chk-/_metadata ``` 3. Resume the statement with `initialSavepointPath` set to that directory. The following example shows only the fields of the request that resume the statement from it: ```json { "spec": { "stopped": false, "startFromSavepoint": { "initialSavepointPath": "s3://my-bucket/checkpoints//chk-" } } } ``` CMF does not check this path, so make sure that it is correct. A job that cannot restore it fails. #### NOTE To move a statement that recovery recreated, for example from an older savepoint, to the latest checkpoint of the job that it lost, update the statement with `startFromSavepoint.initialSavepointPath` set to that checkpoint directory and a new `savepointRedeployNonce`. Its `status.traits.jobId` shows the ID of its new job, so take the ID of the lost job from a record that you kept. ### Resume without state How you resume a statement without state depends on why recovery stopped it: - `no-restore-point` or `restore-point-too-old`: set `upgradeMode` to `stateless` in the request. - `carry-over-replay`: leave `startFromSavepoint` out of the request. The statement is `stateless` already. Without state, the sources of the statement start from their configured startup offsets. A statement that inherited offsets through `sql.tables.initial-offset-from` is the exception: - Its Kafka sources start from the inherited offsets, so the statement processes all input again since then. - To start them from their configured startup offsets instead, recreate the statement without `sql.tables.initial-offset-from`, as described in [Recreate a statement](#cmf-dr-recreate). They can then skip input that arrived during the outage. The following example shows only the fields of the request that resume a statement without state: ```json { "spec": { "stopped": false, "upgradeMode": "stateless" } } ``` The statement keeps `upgradeMode: stateless` after the resume, so its later stops keep no state, and a later recovery that finds no restore point handles it as a `stateless` statement. Leaving `upgradeMode` out of a later update does not change it. To change it, set it, for example to `savepoint`, in a later update. ## Recreate a statement A `FAILED` statement cannot be updated or resumed. To run it again, delete it, and create it again. CMF compiles its plan again, so the new statement starts without the state of the old one. CMF rejects the deletion of a statement that has savepoints, so delete its savepoints first. To recreate a statement: 1. List the savepoints of the statement with `GET /cmf/api/v1/environments/{envName}/statements/{stmtName}/savepoints`. 2. Delete each savepoint with `DELETE /cmf/api/v1/environments/{envName}/statements/{stmtName}/savepoints/{savepointName}?force=true`. 3. Delete the statement with `DELETE /cmf/api/v1/environments/{envName}/statements/{stmtName}`. 4. Create the statement again, as described in [Create Statements with Confluent Manager for Apache Flink](../jobs/sql-statements/create-statements.md#flink-sql-statements-create). #### NOTE When the Flink resource of the statement is gone, deleting its savepoints removes them from CMF, but their files stay in your savepoint storage. For more information, see [Savepoint Management in Confluent Manager for Apache Flink](../jobs/savepoints.md#cmf-savepoints). ## Related content - [Disaster Recovery for Confluent Manager for Apache Flink](overview.md#backup-restore-cmf) - [Troubleshoot Disaster Recovery in Confluent Manager for Apache Flink](troubleshoot.md#cmf-dr-troubleshoot) - [Savepoint Management in Confluent Manager for Apache Flink](../jobs/savepoints.md#cmf-savepoints) - [Manage Statements in Confluent Manager for Apache Flink](../jobs/sql-statements/manage-statements.md#flink-sql-statements-manage)