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. 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.
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.

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:

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.
`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.
`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. 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, 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.

For other symptoms, and for problems whose cause or fix is less obvious, see Troubleshoot Disaster Recovery in Confluent Manager for Apache Flink.

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.
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. 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.
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.
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:

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.
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.
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.
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.

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.

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.

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, from the latest checkpoint, or without state.
`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, from the latest checkpoint, or without state.
`start-from-savepoint-unresolved`	The savepoint that an update set in `startFromSavepoint` no longer resolves.	From a savepoint or from the latest 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, from the latest checkpoint, or without state.
`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.

For example, the status of a statement without a restore point includes the following fields:

{
  "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 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. 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:

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:

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:

{
  "spec": {
    "stopped": false,
    "startFromSavepoint": {
      "savepointName": "orders-sp-41"
    }
  }
}

Resume from the latest checkpoint

Flink stores the checkpoints of a job under <state.checkpoints.dir>/<job_id>/, 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:

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.
In your checkpoint storage, find the checkpoint directory chk-<n> with the highest <n> under <state.checkpoints.dir>/<job_id>/ that contains a _metadata file. A directory without one holds a checkpoint that did not complete. For example, with Amazon S3:
```
aws s3 ls s3://my-bucket/checkpoints/<job_id>/
aws s3 ls s3://my-bucket/checkpoints/<job_id>/chk-<n>/_metadata
```

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:

{
  "spec": {
    "stopped": false,
    "startFromSavepoint": {
      "initialSavepointPath": "s3://my-bucket/checkpoints/<job_id>/chk-<n>"
    }
  }
}

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. 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:

{
  "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:

List the savepoints of the statement with GET /cmf/api/v1/environments/{envName}/statements/{stmtName}/savepoints.
Delete each savepoint with DELETE /cmf/api/v1/environments/{envName}/statements/{stmtName}/savepoints/{savepointName}?force=true.
Delete the statement with DELETE /cmf/api/v1/environments/{envName}/statements/{stmtName}.
Create the statement again, as described in Create Statements with Confluent Manager for Apache Flink.

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.

How Confluent Manager for Apache Flink Recovers Flink SQL Statements

Check your statements after recovery

What startup recovery does with each statement

How CMF chooses the restore point

Limit the age of restore points

Statements that recovery parks

Resume a parked statement

Resume from a savepoint

Resume from the latest checkpoint

Resume without state

Recreate a statement

Related content