<a id="cmf-dr-prepare"></a>

# Prepare for Disaster Recovery in Confluent Manager for Apache Flink

This topic describes what to do before a disaster so that Confluent Manager for Apache Flink® (CMF) can recover your Flink
workloads after the loss of a Kubernetes cluster. You can do some of these steps only while the
cluster that runs your workloads still exists.

For how CMF recovers workloads, and how to recover them, see [Disaster Recovery for Confluent Manager for Apache Flink](overview.md#backup-restore-cmf).

<a id="cmf-dr-backup-metadata"></a>

## Back up CMF metadata

The CMF database is the source of truth for every resource that CMF manages. How you back
it up depends on the database, which is described in [Configure Storage for Confluent Manager for Apache Flink](../installation/storage.md#cmf-storage):

- With a JDBC database (PostgreSQL, Microsoft SQL Server, or IBM Db2), use the backup and restore
  tools of your database. A database that runs outside the Kubernetes cluster that runs your
  Flink workloads also survives the loss of that cluster.
- With the default SQLite database, which CMF stores on a persistent volume, use Kubernetes
  VolumeSnapshots, as described in [Back Up and Restore SQLite Metadata of Confluent Manager for Apache Flink with Volume Snapshots](sqlite-volume-snapshots.md#cmf-dr-volume-snapshots).

A restored database is only as recent as its backup. CMF restores your resources as the backup
records them, so changes that you made after the backup are lost, and CMF does not know about
the resources or savepoints created after the backup. For what to do after such a restore, see
[Verify the recovery](recover.md#cmf-dr-verify).

The metadata does not include the state of your Flink jobs. To keep that state, take savepoints.
For details, see [Take savepoints regularly](#cmf-dr-savepoints).

<a id="cmf-dr-backup-k8s"></a>

### Back up Kubernetes resources and configuration

In addition to the database, keep a copy of the following, which you need to set up a
replacement cluster:

- The encryption key, if encryption is enabled. CMF does not back up the key, and it cannot
  read its database without it. See [Data Encryption in Confluent Manager for Apache Flink](../installation/encryption.md#cmf-encrypt).
- Your Helm values, and the Kubernetes `Secrets` that they reference, such as the kubeconfig
  files of spoke clusters and the credentials for the database and the artifact storage.
- How you installed the Flink Kubernetes Operator: its version, its Helm values, and the
  namespaces that it watches.
- The Kubernetes objects that your workloads need and that CMF does not create, such as
  namespaces, image pull secrets, the service accounts that applications run as, and the
  `Secrets` and `ConfigMaps` that your specs reference.

<a id="cmf-dr-prepare-plans"></a>

## Enable persistence of statements’ compiled plans

A Flink SQL statement runs from a compiled plan, which CMF keeps in a `ConfigMap` on the
Kubernetes cluster that runs the statement. Compiling the statement again after a disaster is no
substitute: a plan compiled again from the same SQL can differ from the original, and then
cannot restore the statement’s savepoints. To recover a statement after the loss of its cluster,
CMF needs a copy of its plan outside the cluster, which it keeps in artifact storage.

Artifact management is off by default, and without it no plan is stored: a statement whose
cluster is lost cannot be recovered, and CMF sets it to `FAILED`. To store plans, enable
artifact management as described in [Enable artifact management](../configure/artifacts.md#cmf-artifacts-enable). The CMF image includes the
filesystem plugins for Amazon S3, Google Cloud Storage, and Azure Blob Storage. For other storage,
see [Filesystem plugins](../configure/artifacts.md#cmf-artifacts-plugins).

CMF stores a copy of a statement’s plan when it deploys the statement, and in the background
copies the plans that are missing: those of statements deployed before CMF 2.5.0 or before you
enabled artifact management, and those whose copy failed. It can copy a plan only while the
Kubernetes cluster that runs the statement still exists. The plans of interactive `SELECT`
statements are not stored.

#### IMPORTANT
Enable artifact management, and wait until CMF has copied every plan, while the Kubernetes
clusters that run your statements still exist. See [Check that every statement is protected](#cmf-dr-check-protection).

For where CMF stores the plans, and how to change the base path without losing them, see
[Compiled plans of SQL statements](../configure/artifacts.md#cmf-artifacts-statement-plans). A stored plan that is changed or deleted cannot recover its
statement, so restrict write access to the plans, as described in
[Restrict write access to stored plans](../configure/artifacts.md#cmf-artifacts-plan-write-access).

<a id="cmf-dr-check-protection"></a>

## Check that every statement is protected

Enable CMF metrics with the `cmf.metrics.enabled` Helm value, and check the following gauges
on the metrics endpoint, `/actuator/prometheus` on port `9404` by default. The metrics port
has no authentication and no TLS, so restrict access to it, for example with a Kubernetes
`NetworkPolicy`. CMF updates the gauges each time it copies missing plans, so they appear only
after it first does so.

`cmf_statements_durable_plan_waiting`
: The number of statements whose plan CMF has not copied yet. Wait until it is `0`. If it
  stays above `0`, CMF cannot copy some plans, and logs a warning that says why, such as
  `Could not read plan ConfigMaps` or `The artifact store failed a write`.

`cmf_statements_dr_unprotected`
: The number of statements that cannot be recovered after the loss of their cluster, or whose
  state might not survive it, by `reason`. A statement is counted under each reason that applies
  to it:
  <br/>
  - `artifacts-disabled`: artifact management is off, so no plan is stored. Enable it.
  - `plan-configmap-missing`: the statement’s plan `ConfigMap` is missing or holds no plan,
    and no copy of the plan is stored.
  - `plan-not-decryptable`: the plan in the statement’s `ConfigMap` does not decrypt with the
    statement’s key, so it is not stored.
  - `compute-pool-missing`: the statement’s compute pool does not exist, so CMF can neither
    deploy nor recover the statement.
  - `state-storage-unset`: the statement’s checkpoint or savepoint directory is not set. Set it
    as described in [Keep checkpoints and savepoints on durable storage](#cmf-dr-state-storage).
  - `state-storage-local`: the statement’s checkpoint or savepoint directory has no scheme, and
    `fs.default-scheme` sets none, so it is on the local file system of its Flink pods. Set it
    as described in [Keep checkpoints and savepoints on durable storage](#cmf-dr-state-storage).
  <br/>
  With `plan-configmap-missing` or `plan-not-decryptable`, CMF has no plan that it can
  store: the `ConfigMap` was removed or changed outside CMF, or it was lost with its cluster
  before CMF stored a copy. To protect the statement, recreate it as described in
  [Recreate a statement](statements.md#cmf-dr-recreate), which compiles its plan again and starts it without its state.

<a id="cmf-dr-state-storage"></a>

## Keep checkpoints and savepoints on durable storage

A recovered statement or application can restore its state only from checkpoint and savepoint
storage that survived the disaster. Set `state.checkpoints.dir` and `state.savepoints.dir` to
durable storage outside the Kubernetes cluster that a replacement cluster can reach, such as
`s3://`, `gs://`, `abfs://`, or `hdfs://` locations:

- For statements, set them in the `clusterSpec.flinkConfiguration` of the compute pool, or
  for all compute pools of an environment in its `computePoolDefaults`.
  - If statements are running on the compute pool, stop them, update the compute pool, and
    resume them, as described in [Update a compute pool](../configure/compute-pools.md#cmf-compute-pools-update).
  - Stopping and resuming a statement loses its state if it keeps the state on the local file
    system of its Flink pods, or if it is `stateless`, which is the default on a compute pool
    that does not set both directories. To keep its state, take a savepoint of it with
    `spec.path` set to durable storage before you stop it, and resume it from that savepoint,
    as described in [Savepoint Management in Confluent Manager for Apache Flink](../jobs/savepoints.md#cmf-savepoints).
- For applications, set them in the application’s `spec.flinkConfiguration`.

For more information, see [Checkpointing with Confluent Manager for Apache Flink](../jobs/configure/checkpointing.md#cmf-checkpointing).

Setting both directories also makes `savepoint` the default `upgradeMode` of the statements on
the compute pool, as described in [Manage Statements in Confluent Manager for Apache Flink](../jobs/sql-statements/manage-statements.md#flink-sql-statements-manage). Recovery stops a `savepoint`
statement that has no savepoint instead of starting it without state, so take savepoints regularly,
as described in [Take savepoints regularly](#cmf-dr-savepoints).

#### NOTE
CMF logs a warning for each statement whose checkpoint or savepoint directory is not set or is
on the local file system of its Flink pods, so that you can find the statements to move to durable
storage. It also counts them in the `cmf_statements_dr_unprotected` metric, as described in
[Check that every statement is protected](#cmf-dr-check-protection). A `file://` directory is logged but not counted, because it can
be a shared mount outside the cluster. Batch statements are not checked.

<a id="cmf-dr-savepoints"></a>

## Take savepoints regularly

When CMF recreates a statement, it restores the statement’s latest savepoint, as described in
[How CMF chooses the restore point](statements.md#cmf-dr-restore-point). To recover with recent state, schedule savepoints for every
statement whose state matters. For details, see [Schedule periodic savepoints](../jobs/savepoints.md#cmf-periodic-savepoints). The age of the
savepoint decides how much input a recovered statement processes again, and CMF cleans up only
scheduled savepoints, so a manual one can be months old.

#### IMPORTANT
Schedule savepoints for statements whose `upgradeMode` is `savepoint` or `last-state`,
and for statements that start their Kafka sources from the offsets of another statement through
`sql.tables.initial-offset-from`. Recovery stops such a statement when it has no savepoint, as
described in [Statements that recovery parks](statements.md#cmf-dr-parked).

CMF does not choose a savepoint for an application: a recovered application restores the
savepoint that its spec names, if any, and otherwise starts without state. To start a stateful
application from a savepoint that you choose instead, see [Recovery settings](overview.md#cmf-dr-recovery-settings).

## Rehearse recovery

Rehearse the procedure in [Recover Workloads onto a Replacement Cluster with Confluent Manager for Apache Flink](recover.md#cmf-dr-recover) on a test installation, so that you know how
long recovery takes and how your workloads come back before you need it.

## Related content

- [Disaster Recovery for Confluent Manager for Apache Flink](overview.md#backup-restore-cmf)
- [Back Up and Restore SQLite Metadata of Confluent Manager for Apache Flink with Volume Snapshots](sqlite-volume-snapshots.md#cmf-dr-volume-snapshots)
- [Configure Storage for Confluent Manager for Apache Flink](../installation/storage.md#cmf-storage)
- [Data Encryption in Confluent Manager for Apache Flink](../installation/encryption.md#cmf-encrypt)
- [Manage Artifacts in Confluent Manager for Apache Flink](../configure/artifacts.md#cmf-artifacts)
- [Savepoint Management in Confluent Manager for Apache Flink](../jobs/savepoints.md#cmf-savepoints)
- [Checkpointing with Confluent Manager for Apache Flink](../jobs/configure/checkpointing.md#cmf-checkpointing)