# Savepoint Management in Confluent Manager for Apache Flink A Savepoint resource in Confluent Manager for Apache Flink (CMF) is a managed reference to an Apache Flink® savepoint, which is an image of the execution state of a streaming job. The resource tracks the lifecycle, location, and metadata of the Flink savepoint. You can manage Savepoint resources for both CMF applications and statements with the CMF API, starting with CMF 2.1.0. Starting with CMF 2.4.0, you can also configure savepoints to be triggered on a recurring schedule, with retention policies, blackout windows, and environment-wide defaults. See [Schedule periodic savepoints](#cmf-periodic-savepoints). ## Savepoints versus checkpoints Flink creates *checkpoints* automatically to support fault-tolerant recovery. If a running job fails, Flink restarts it from the last completed checkpoint without operator intervention. Checkpoints are managed by the Flink runtime, have short retention, and are tied to a specific running job. *Savepoints* are triggered explicitly, either by you or by CMF. A savepoint is a self-contained snapshot that is retained until you delete it and that can be used to restart the same job or a different job. Use savepoints for planned operations such as the following: - Upgrading a job to a new Flink version or application image. - Migrating a job between clusters or environments. - Preserving state before decommissioning a job. - Forking a job for testing, blue/green deployment, or A/B rollout. - Establishing known-good recovery points for disaster recovery. For more information on the underlying Flink concepts, see the [Apache Flink documentation on checkpoints and savepoints](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/checkpoints_vs_savepoints/). ## Savepoint lifecycle states A CMF savepoint can have the following states in `status.state`: `TRIGGER_PENDING` : The request to trigger the savepoint has been accepted and is waiting to be processed. `IN_PROGRESS` : The savepoint operation is currently running. `COMPLETED` : Savepoint creation was successful. The final location is available in `status.path`. `FAILED` : Savepoint creation failed. Check `status.error` for details about the failure. You can check the number of retries in `status.failures`. `ABANDONED` : The savepoint operation was abandoned, potentially due to a job shutdown or other interruption. `DELETING` : CMF accepted a request to delete the savepoint. The deletion is in progress. ## Savepoint formats When triggering a savepoint, you can specify the format in `spec.formatType`: `CANONICAL` : The default. A state-backend-independent format that can be restored to a job using any state backend. Use `CANONICAL` if you might switch state backends (for example, from `hashmap` to `rocksdb`), or if you are not sure which format to pick. `NATIVE` : The state backend’s native binary format. Faster to create and typically smaller than a `CANONICAL` savepoint for large state, but can only be restored to a job using the same state backend that created it. Use `NATIVE` when savepoint creation time matters and you do not plan to change state backends. #### NOTE You cannot create a `CANONICAL` savepoint for a job that uses asynchronous state (operators that call `enableAsyncState()`). For more information on formats, see the [Apache Flink documentation on savepoint formats](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/#savepoint-format). ## Default savepoint configuration values The following `spec` fields have default values if not explicitly provided: - `spec.formatType`: Defaults to `CANONICAL` if not specified. - `spec.backoffLimit`: Defaults to `-1`, which means that CMF will retry a failed savepoint operation indefinitely until it succeeds or is manually abandoned. Unlimited retries let a savepoint attempt survive transient issues (for example, a temporary blob store outage) rather than dropping silently. If you need bounded retries—for example, on a manual savepoint tied to a deploy step—set `backoffLimit` to a positive integer. ## Attached versus Detached savepoints CMF manages two kinds of savepoints: Attached savepoints : These are tied to a specific Flink application or statement. You can start any Flink application or statement from its attached savepoint. Detached savepoints : These are standalone resources not tied to a specific running job. Use them to import savepoints from external systems, such as open source Flink, to preserve savepoints from deleted jobs, or to share a savepoint between Flink applications—for example, to start a new application from a savepoint originally taken by another application. Only Flink applications support starting from a detached savepoint. Statements can start from a savepoint attached to the same statement or from a raw savepoint path. The following table summarizes which savepoint operations are supported for applications and statements: | Operation | Application | Statement | |----------------------------------------------------|---------------|-------------| | Trigger a new attached savepoint | Yes | Yes | | Detach an attached savepoint | Yes | No | | Start the same job from its own attached savepoint | Yes | Yes | | Start a job from any detached savepoint | Yes | No | | Start a job from a raw `initialSavepointPath` | Yes | Yes | ## Manage savepoints with APIs The CMF REST API provides operations to trigger, import, list, detach, and delete savepoints. For a full list of REST APIs, see [REST APIs for Confluent Manager for Apache Flink](../clients-api/rest.md#af-rest-api). ### Trigger a new attached savepoint You can manually trigger a new savepoint for a running Flink application or statement. **For Applications:** ```none POST /cmf/api/v1/environments/{env}/applications/{appName}/savepoints ``` **For Statements:** ```none POST /cmf/api/v1/environments/{env}/statements/{stmtName}/savepoints ``` **Request Body:** ```json { "apiVersion": "cmf.confluent.io/v1", "kind": "Savepoint", "metadata": { "name": "my-manual-savepoint-1" }, "spec": { "path": "s3://my-bucket/custom-path/", "formatType": "CANONICAL", "backoffLimit": 0 } } ``` **Field Explanations:** - `metadata.name` (Optional): A user-provided name for the savepoint. If the name is not provided, CMF will generate one based on the parent resource. - `spec.path` (Optional): The directory where the savepoint data should be stored. If not provided, the Flink cluster’s default configured savepoint directory will be used. - `spec.formatType` (Optional): Specifies the savepoint format. Can be `CANONICAL` or `NATIVE`. Defaults to `CANONICAL`. For more information, see [Savepoint Formats](#cmf-savepoint-formats). - `spec.backoffLimit` (Optional): The maximum number of retries if the savepoint operation fails. `0` means no retries. `-1` means unlimited retries. Defaults to `-1`. For more information, see [Default Values](#cmf-savepoint-defaults). The trigger request returns immediately. The savepoint in the response does not have a `status.state` yet. The state is set when the Kubernetes Operator starts processing the savepoint. To confirm the savepoint completed, `GET` the savepoint resource and wait for `status.state` to reach `COMPLETED`. The completed savepoint’s storage path is returned in `status.path`. If the state reaches `FAILED` instead, inspect `status.error` and `status.failures` for details. ### Adopt savepoints from the Kubernetes Operator Besides the savepoints you manually trigger, CMF automatically *adopts* savepoints created by the Confluent Platform for Apache Flink Kubernetes Operator (FKO). The FKO takes a savepoint when it upgrades or suspends a job. CMF detects each FKO-created savepoint and creates a corresponding CMF Savepoint resource. Adopted savepoints appear in the list of attached savepoints for the Flink application or statement, and you can manage them like any other attached savepoint. Adopted upgrade and suspend savepoints are tagged with `status.source: UPGRADE`. Savepoints produced by the FKO periodic savepoint interval are tagged `status.source: SCHEDULE`. For the full list of savepoint sources, see [Filter and bulk-delete savepoints by source](#cmf-periodic-savepoints-source). ### Import an existing savepoint (detached) To import an existing savepoint, for example one produced by open source Flink, you create a detached savepoint. This operation “registers” the existing savepoint with CMF. The state of a detached savepoint is always `COMPLETED`. ```none POST /cmf/api/v1/detached-savepoints ``` **Request Body:** ```json { "apiVersion": "cmf.confluent.io/v1", "kind": "DetachedSavepoint", "metadata": { "name": "my-detached-savepoint-1" }, "spec": { "path": "s3://flink/stateful-flink/checkpoints" } } ``` **Field Explanations:** - `metadata.name` (Required): The name of the savepoint. This must be globally unique. - `spec.path` (Required): The full, exact path to the completed savepoint data. ### List and read savepoints #### List Attached Savepoints **For Applications:** ```none GET /cmf/api/v1/environments/{env}/applications/{appName}/savepoints ``` **For Statements:** ```none GET /cmf/api/v1/environments/{env}/statements/{stmtName}/savepoints ``` #### List Detached Savepoints ```none GET /cmf/api/v1/detached-savepoints ``` #### Read a Single Savepoint **For Applications:** ```none GET /cmf/api/v1/environments/{env}/applications/{appName}/savepoints/{savepointName} ``` **For Detached Savepoints:** ```none GET /cmf/api/v1/detached-savepoints/{savepointName} ``` ### Detach a savepoint Converting an attached savepoint to a detached savepoint preserves it before deleting the parent application or allows sharing it with other jobs. This can be done only for savepoints in the `COMPLETED` state. **For Applications only:** ```none POST /cmf/api/v1/environments/{env}/applications/{appName}/savepoints/{savepointName}/detach ``` #### NOTE Detaching is only supported for FlinkApplications. Savepoints attached to a Statement cannot be detached. ### Delete a savepoint Attached savepoints delete physical data from storage when possible. Detached savepoints only remove metadata from CMF. #### Delete an attached savepoint Delete an attached savepoint for an application: ```none DELETE /cmf/api/v1/environments/{env}/applications/{appName}/savepoints/{savepointName}?force=(true|false) ``` Delete an attached savepoint for a statement: ```none DELETE /cmf/api/v1/environments/{env}/statements/{stmtName}/savepoints/{savepointName}?force=(true|false) ``` #### IMPORTANT This endpoint performs a best-effort deletion of the physical savepoint data from the blob store. CMF does not guarantee that the underlying savepoint files are removed, and orphaned data can remain if the delete step fails. The corresponding Flink job must be running for the deletion to be processed. If the job is not running, CMF rejects the request with `HTTP 409` and keeps the Savepoint resource. Retry the delete after the job is running again, or use `force=true`. #### WARNING Using `force=true` immediately deletes the Savepoint resource from CMF and Kubernetes regardless of whether the physical data was successfully removed. This can leave orphaned savepoint data in your blob store; use with caution. #### Delete a detached savepoint ```none DELETE /cmf/api/v1/detached-savepoints/{savepointName} ``` This call only deletes the savepoint’s metadata from the CMF database. It does not delete the underlying physical savepoint data from the blob store. The user is expected to clean up the physical data manually if needed. ## Start a job from a savepoint ### Start applications To start a Flink application from a savepoint, you have two options: 1. Use create or update with the `application` endpoint. When creating (`POST`) or updating (`PUT`) a Flink application, you can specify the `startFromSavepoint` field in the spec. You must specify one of the following mutually exclusive fields: - `savepointName`: The name of a CMF Savepoint resource (either attached or detached). - `uid`: The UID of a CMF Savepoint resource. To start from a raw path to a savepoint that is not managed by CMF, such as `s3://.../sp-1`, set `spec.job.initialSavepointPath` instead of `startFromSavepoint`. You cannot combine `spec.job.initialSavepointPath` with `startFromSavepoint`. All properties in the [Flink Kubernetes Operator JobSpec reference](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-1.15/docs/custom-resource/reference/#jobspec) are supported for specifying in a Flink Application. For example, to set `allowNonRestoredState` when starting from a savepoint: ```yaml apiVersion: cmf.confluent.io/v1 kind: FlinkApplication metadata: name: fraud-detection-from-savepoint spec: image: confluentinc/cp-flink:1.19.1-cp1 flinkVersion: v1_19 flinkConfiguration: taskmanager.numberOfTaskSlots: "2" metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory metrics.reporter.prom.port: "9249-9250" serviceAccount: flink jobManager: resource: memory: "1024m" cpu: 1 taskManager: resource: memory: "1024m" cpu: 1 job: jarURI: "local:///opt/flink/examples/streaming/StateMachineExample.jar" state: running parallelism: 2 upgradeMode: savepoint allowNonRestoredState: true startFromSavepoint: savepointName: "fraud-detection-savepoint-1" ``` The `savepointRedeployNonce` field controls whether an update to `startFromSavepoint` triggers a redeploy. If you update `startFromSavepoint` on an already-deployed application without also changing `savepointRedeployNonce` to a new non-null value, CMF does not redeploy the job to re-apply the savepoint. This matches standard Confluent Platform for Apache Flink Kubernetes Operator behavior. 2. Start an application with a savepoint. You can use the `start` endpoint with a query parameter to force a start from a specific savepoint. ```none POST /cmf/api/v1/environments/{envName}/applications/{appName}/start?startFromSavepointUid={uid} ``` This operation always forces the application to start from the specified savepoint UID, even if the application’s `spec` already references the same savepoint. This is done by automatically updating the deployment nonce to trigger a new deploy. ### Start statements To start a statement from a savepoint, set the `startFromSavepoint` field when you update the statement with `PUT`. CMF rejects `startFromSavepoint` when you create a statement. Specify exactly one of the following fields: - `savepointName` or `uid`: References a savepoint that is attached to the same statement. - `initialSavepointPath`: A raw path to a savepoint that is not managed by CMF, such as `s3://.../sp-1`. Starting a statement from a detached savepoint is not supported. CMF rejects a request that references a detached savepoint by name or UID. Starting with CMF 2.4.3, CMF uses an `initialSavepointPath` as given, even if it is the path of a detached savepoint. Earlier 2.4 releases reject such a path. To trigger redeployment of a statement from a savepoint, change the `savepointRedeployNonce` field in the statement’s `spec.startFromSavepoint` to a different non-null value. A changed value is required to trigger a new deployment. ## Schedule periodic savepoints You can configure CMF to trigger savepoints on a recurring schedule and to automatically clean up old scheduled savepoints. This keeps a rolling history of recovery points for a Flink application or statement without triggering each savepoint by hand. Periodic savepoints support use cases such as disaster recovery and point-in-time recovery. A regularly maintained set of recent savepoints gives you known-good states to restart a job from after data corruption, a bad deployment, or the loss of a cluster. For more about backup and restore in CMF, see [Disaster Recovery using VolumeSnapshots with Confluent Manager for Apache Flink](../disaster-recovery.md#backup-restore-cmf). #### NOTE Scheduled savepoints are [attached savepoints](#cmf-savepoints-attached-vs-detached). They follow the same lifecycle states, formats, and deletion rules described earlier on this page. Schedule configuration is API-only in this release. The Confluent CLI does not manage schedules. The CMF UI shows schedule status and lets you pin or unpin individual savepoints, but you cannot author or edit a schedule from the UI. ### Quick start Before you begin: - A savepoint storage location must be reachable by the Flink cluster. For an application, set `state.savepoints.dir` under `spec.flinkConfiguration` on the application or configure a cluster-wide default. For a statement, set `state.savepoints.dir` in the compute pool’s `spec.clusterSpec.flinkConfiguration` or in the environment’s `computePoolDefaults`. Setting it in the statement’s `spec.flinkConfiguration` has no effect. This is the same precondition as manual savepoints. Add a `savepointSchedule` to the `spec` of an application or statement. The following example creates a scheduled savepoint every six hours and keeps the ten most recent: ```yaml apiVersion: cmf.confluent.io/v1 kind: FlinkApplication metadata: name: fraud-detection spec: # ... other application spec fields ... flinkConfiguration: state.savepoints.dir: s3://my-bucket/savepoints/ savepointSchedule: cronExpression: "0 0 */6 * * *" retention: keepLast: 10 ``` The same `savepointSchedule` block is valid in a statement `spec`. Only detached non-`SELECT` statements are scheduled. A `SELECT` statement ignores the field. For the full list of fields, see [Configure a schedule](#cmf-periodic-savepoints-config). ### How scheduling works When a resource has a `spec.savepointSchedule`, CMF does the following: 1. Registers the schedule and computes the next trigger time from the cron expression and time zone. 2. At each trigger, creates a new attached Savepoint resource with `status.source` set to `SCHEDULE`, unless the schedule is paused, the trigger falls inside a blackout window, the parent job is not running, or a previous scheduled savepoint is still in flight. 3. Periodically applies the retention policy, deleting scheduled savepoints that no rule keeps. CMF retention runs only against scheduled savepoints created by a CMF schedule. Manual and upgrade savepoints, and any savepoint marked as pinned, are always kept until you delete them. Savepoints adopted from the FKO periodic interval (`FKO_MANAGED`; see [Interoperability with the Confluent Platform for Apache Flink Kubernetes Operator](#cmf-periodic-savepoints-fko)) also have `status.source: SCHEDULE` but are governed by the FKO’s own retention configuration, not by CMF retention rules. If a resource uses the Confluent Platform for Apache Flink Kubernetes Operator (FKO) periodic savepoint interval instead of a `savepointSchedule`, CMF adopts the FKO-triggered savepoints under the same `SCHEDULE` source. See [Interoperability with the Confluent Platform for Apache Flink Kubernetes Operator](#cmf-periodic-savepoints-fko). ### Monitor a schedule Every `FlinkApplication` and `Statement` with a schedule surfaces a read-only `status.savepointSchedule` object in its GET response. In the CMF UI, this appears as a status card above the savepoints table on the resource details page.  The status object contains the following fields: - `state`: The current lifecycle state of the schedule. One of `ACTIVE`, `PAUSED`, `FKO_MANAGED`, or `DISABLED`. See the table below. - `nextTriggerTime`: The next time the schedule is evaluated for a trigger. - `lastTriggerTime`: The last time CMF attempted to create a savepoint on the schedule. Skipped fires (blackout, resource not ready, prior savepoint still in flight) do not update this field. - `lastCompletedTime`: The time the most recent savepoint for this resource completed successfully, across all sources. This is the value the freshness SLA check measures against. - `lastErrorMessage`: The most recent trigger failure observed, or empty when the schedule is healthy. | State | Meaning | What to do | |---------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `ACTIVE` | The schedule fires on its cron expression. | Nothing. This is the healthy state. | | `PAUSED` | The schedule is registered but no new savepoints are triggered because
`paused: true` is set on the resource, on the environment default, or
on both. | To resume triggers, clear `paused` everywhere it is set. The resource
cannot override `paused: true` on the environment default. After you
clear the flag on the environment default, update each resource to
apply the change. Retention cleanup continues regardless of pause
state. | | `FKO_MANAGED` | No CMF schedule row exists for this resource, but its
`flinkConfiguration` sets
`kubernetes.operator.periodic.savepoint.interval`. The FKO is
triggering the savepoints; CMF adopts them as `SCHEDULE` source. | Use CMF scheduling or the FKO interval, not both. See
[Interoperability with the Confluent Platform for Apache Flink Kubernetes Operator](#cmf-periodic-savepoints-fko). | | `DISABLED` | CMF stopped the schedule after five consecutive trigger failures.
`lastErrorMessage` contains the last failure reason. | Edit any trigger-group field of the schedule (for example,
`cronExpression`, `timezone`, `paused`, `jitterSeconds`,
`blackoutWindows`, or `catchUpOnRestart`) to reset the failure
counter. A retention-only edit does not re-enable a disabled schedule. | To find the source of any savepoint, look at `status.source` on the savepoint itself. The CMF UI surfaces this as a Source column in the attached savepoints table:  ### Configure a schedule The following fields are available in `spec.savepointSchedule`: | Field | Type | Default | Description | |---------------------------|---------|----------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `cronExpression` | string | Required | Spring six-field cron expression that controls when savepoints fire.
The minimum interval between consecutive triggers is five minutes.
See [Cron expressions](#cmf-periodic-savepoints-cron). | | `timezone` | string | `UTC` | IANA time zone ID used to evaluate the cron expression. For example,
`America/New_York`. | | `paused` | boolean | `false` | When `true`, the schedule is registered but no new savepoints are
triggered. Retention still runs. | | `catchUpOnRestart` | boolean | `false` | When `true`, if CMF restarts after missing at least one scheduled
trigger, CMF fires exactly one catch-up trigger on startup. Never
fires when no previous trigger exists. | | `jitterSeconds` | integer | auto | Upper bound, in seconds, for a random offset added to each trigger
time. See [Jitter](#cmf-periodic-savepoints-jitter). | | `formatType` | string | `CANONICAL` | Format used for every scheduled savepoint. One of `CANONICAL` or
`NATIVE`. See [Savepoint formats](#cmf-savepoint-formats). | | `blackoutWindows` | array | empty | Time ranges during which scheduled triggers are suppressed. See
[Blackout windows](#cmf-periodic-savepoints-blackout). | | `freshnessThresholdHours` | integer | `0` (disabled) | Maximum acceptable gap, in hours, between successful savepoints. When
exceeded, CMF logs a warning. See
[Freshness SLA](#cmf-periodic-savepoints-sla). | | `retention` | object | none | Retention policy for scheduled savepoints. If omitted, scheduled
savepoints accumulate without automatic cleanup. See
[Retention policy](#cmf-periodic-savepoints-retention). | #### Cron expressions CMF uses Spring’s six-field cron format. The fields, in order, are: second, minute, hour, day-of-month, month, and day-of-week. Spring macros such as `@hourly`, `@daily`, and `@weekly` are also accepted. Some examples: - `"0 0 */6 * * *"`: every six hours, on the hour. - `"0 30 2 * * MON"`: at 02:30 every Monday. - `"0 0 0 * * *"`: at midnight every day. The following rules apply: - The expression must have six fields. Five-field expressions (Unix or Quartz style) are rejected with an HTTP 400 error that asks you to add the seconds field. - The interval between consecutive triggers must be at least five minutes. Shorter intervals are rejected with an HTTP 400 error. - The expression is evaluated in the schedule’s `timezone`. For zones that observe daylight saving time, a trigger can fire twice (when clocks move back) or be skipped (when clocks move forward) at the transition. Use `UTC` if you need every interval to be exactly the same length. #### Jitter Jitter adds a small random offset to each trigger time so that many resources sharing the same schedule (for example, through an environment default) do not all trigger savepoints at the same instant. CMF assigns each schedule a fixed offset within the jitter window and reuses it across restarts, so a given resource always triggers at the same point in its window. Jitter does not shift the cron clock: the next trigger time is always computed from the original scheduled time. Control jitter with `jitterSeconds`: - Omit the field, or set it to `null`, to let CMF choose a bound automatically: `min(300, cronIntervalSeconds / 2)`. - Set it to a positive value to use that value, in seconds, as the upper bound. Setting `jitterSeconds: 0` behaves the same as omitting the field: CMF still computes an offset from the automatic bound. To minimize jitter for a single, standalone schedule, set `jitterSeconds: 1`. CMF chooses the offset once, when it first registers the schedule. Changing `jitterSeconds` or `cronExpression` later does not choose a new offset. To apply a new bound, remove the schedule from the resource, and then add it again. #### Blackout windows Blackout windows suppress scheduled triggers during recurring time ranges, for example during peak load hours or a nightly batch load. Triggers missed during a blackout are skipped, not retried. Each entry in `blackoutWindows` has the following fields: - `cronExpression` (Required): A six-field Spring cron expression whose firing times mark the start of each blackout window. - `durationMinutes` (Optional): How long the window lasts, in minutes, from the cron fire time. Defaults to `60`. - `timezone` (Optional): The IANA time zone for this window’s cron expression. Defaults to the schedule’s `timezone`. The following example suppresses scheduled savepoints for one hour starting at 09:00 on weekdays: ```yaml savepointSchedule: cronExpression: "0 0 */2 * * *" blackoutWindows: - cronExpression: "0 0 9 * * MON-FRI" durationMinutes: 60 ``` If any blackout window is malformed, CMF treats the current time as in-blackout for that schedule. This is a fail-safe: an unparseable window suppresses triggers rather than fires them. #### Freshness SLA Set `freshnessThresholdHours` to the maximum acceptable gap, in hours, between successful scheduled savepoints. When the gap is exceeded, CMF emits a `SAVEPOINT_SLA_VIOLATION` warning to the CMF server log. The check measures freshness against `lastCompletedTime` (the time the most recent savepoint for the resource completed successfully, across all sources), not against trigger activity. A value of `0` (the default) disables the check. #### NOTE `SAVEPOINT_SLA_VIOLATION` is a server log line only. It is not exposed through the CMF events API or the savepoints API. ### Retention policy The retention policy decides which scheduled savepoints CMF keeps. It uses a grandfather-father-son (GFS) model: a savepoint survives cleanup if any configured rule keeps it. Set one or more of the following rules in `spec.savepointSchedule.retention`. If you include `retention`, you must set at least one rule; otherwise, CMF rejects the request with an HTTP 400. If you omit `retention` entirely, scheduled savepoints accumulate without automatic cleanup. | Rule | Type | Semantics | |---------------|--------------------|-----------------------------------------------------------------------------------------------------------------------------------| | `keepLast` | integer, minimum 1 | Keep the N most recently completed scheduled savepoints. This is an
absolute floor. | | `keepWithin` | ISO-8601 duration | Keep every completed scheduled savepoint taken within this duration of
now. Examples: `"PT24H"` (24 hours), `"P7D"` (7 days). | | `keepHourly` | integer, minimum 1 | Keep the latest completed scheduled savepoint per calendar hour, for
the most recent N hours. | | `keepDaily` | integer, minimum 1 | Keep the latest completed scheduled savepoint per calendar day, for
the most recent N days. | | `keepWeekly` | integer, minimum 1 | Keep the latest completed scheduled savepoint per ISO week (Monday
start), for the most recent N weeks. | | `keepMonthly` | integer, minimum 1 | Keep the latest completed scheduled savepoint per calendar month, for
the most recent N months. | | `keepYearly` | integer, minimum 1 | Keep the latest completed scheduled savepoint per calendar year, for
the most recent N years. | Calendar buckets (hourly, daily, weekly, monthly, and yearly) are aligned in the schedule’s `timezone`. Manual savepoints, upgrade savepoints, and [pinned](#cmf-periodic-savepoints-pin) savepoints are never removed by retention. The following policy keeps at least the five most recent scheduled savepoints, plus one per day for seven days, plus one per week for four weeks: ```yaml retention: keepLast: 5 keepDaily: 7 keepWeekly: 4 ``` Retention runs on a fixed-delay task, once every 60 minutes by default. The task removes eligible savepoints one at a time; if the backing Flink cluster is unavailable, CMF logs a warning for that savepoint and continues with the next candidate. CMF administrators can tune the cleanup interval; see [Automatic cleanup of failed periodic savepoints](#cmf-periodic-savepoints-cleanup). ### Pin a savepoint To protect an individual scheduled savepoint from automatic retention, set `spec.pinned` to `true` on the Savepoint resource. A pinned savepoint is excluded from the retention candidate set and must be deleted explicitly. In the CMF UI, pin and unpin actions appear on the row menu for each scheduled savepoint. The action is not shown for manual or upgrade savepoints, which are already exempt from cleanup.  To pin or unpin a savepoint through the API, GET the savepoint, set `spec.pinned`, and send the full resource back with PUT. ```none PUT /cmf/api/v1/environments/{env}/applications/{appName}/savepoints/{savepointName} ``` ```none PUT /cmf/api/v1/environments/{env}/statements/{stmtName}/savepoints/{savepointName} ``` Example request body: ```json { "apiVersion": "cmf.confluent.io/v1", "kind": "Savepoint", "metadata": { "name": "fraud-detection-scheduled-20260713120000-a1b2c3d4" }, "spec": { "pinned": true, "formatType": "CANONICAL", "backoffLimit": -1 } } ``` Only `spec.pinned` is mutable on this endpoint, and the request must include it. If the body sets `spec.path`, `spec.backoffLimit`, or `spec.formatType` to a value that differs from the stored value, the request is rejected with `HTTP 400`. Omitted fields, `backoffLimit: -1`, and `formatType: CANONICAL` are always accepted. If `metadata.name` is present in the body, it must match the savepoint name in the URL. ### Set defaults for an environment To apply the same schedule to every application and every detached statement in an environment, set a default schedule on the environment. Applications and statements read the default from different fields. Both fields use the same shape as `spec.savepointSchedule`: - `flinkApplicationDefaults.spec.savepointSchedule` applies to applications. - `statementDefaults.detached.savepointSchedule` applies to detached statements. Interactive (`SELECT`) statements never inherit an environment schedule. CMF rejects `statementDefaults.interactive.savepointSchedule`. To use one schedule for applications and statements, set it in both fields: ```json { "metadata": { "name": "production" }, "kubernetesNamespace": "flink", "flinkApplicationDefaults": { "spec": { "savepointSchedule": { "cronExpression": "0 0 */6 * * *", "retention": { "keepLast": 5 } } } }, "statementDefaults": { "detached": { "savepointSchedule": { "cronExpression": "0 0 */6 * * *", "retention": { "keepLast": 5 } } } } } ``` When a resource has both an environment default and a resource-level `savepointSchedule`, CMF merges the two: - Scalar fields (`cronExpression`, `timezone`, `catchUpOnRestart`, `jitterSeconds`, `freshnessThresholdHours`, `formatType`): the environment value wins per-field when it is not null. A null field on the environment default falls back to the resource value. - `paused` uses OR semantics: if either the environment default or the resource sets `paused: true`, the effective schedule is paused. - `retention` and `blackoutWindows`: when the environment default sets these, the environment’s value replaces the resource’s value as a whole unit rather than merging field by field. If the environment default omits either field, the resource’s value applies. #### IMPORTANT Environment default changes are applied lazily. CMF re-materializes a resource’s schedule on the next create or update of that resource, not immediately when you change the environment default. Removing a default does not clean up already-materialized schedule rows. To roll out a new default across an environment, redeploy affected resources. ### Filter and bulk-delete savepoints by source Every Savepoint has a read-only `status.source` that records how it was created: - `MANUAL`: You triggered it explicitly through the REST API. - `UPGRADE`: CMF adopted it from an FKO upgrade or suspend snapshot. - `SCHEDULE`: CMF created it from a periodic savepoint schedule, or adopted it from the FKO periodic savepoint interval when the resource is `FKO_MANAGED` (see [Interoperability with the Confluent Platform for Apache Flink Kubernetes Operator](#cmf-periodic-savepoints-fko)). To list only savepoints with a given source, add a `filter` query parameter with a `source` expression to the list endpoint. Use this as a dry run before a bulk delete: ```none GET /cmf/api/v1/environments/{env}/applications/{appName}/savepoints?filter=source=SCHEDULE ``` ```none GET /cmf/api/v1/environments/{env}/statements/{stmtName}/savepoints?filter=source=SCHEDULE ``` To delete many savepoints in one request, use the bulk delete endpoint. The following call deletes every scheduled savepoint older than 30 days: ```none DELETE /cmf/api/v1/environments/{env}/applications/{appName}/savepoints?source=SCHEDULE&olderThanDays=30 ``` ```none DELETE /cmf/api/v1/environments/{env}/statements/{stmtName}/savepoints?source=SCHEDULE&olderThanDays=30 ``` Query parameters: - `source` (Required): Only savepoints with this source are deleted. One of `MANUAL`, `UPGRADE`, or `SCHEDULE`. No single value deletes every source at once; to clear multiple sources, issue one call per source. - `olderThanDays` (Optional): Delete only savepoints older than this many days. Mutually exclusive with `olderThanHours`. - `olderThanHours` (Optional): Delete only savepoints older than this many hours. Mutually exclusive with `olderThanDays`. - `force` (Optional): Force-delete even if the backing Flink cluster is unavailable. Defaults to `false`. As with single-savepoint deletion, `force=true` can leave orphaned savepoint data in your blob store. Pinned savepoints are excluded from the bulk-delete candidate set. They do not appear in `deletedCount` or `failures`. The response reports how many savepoints were deleted and lists any that failed: ```json { "deletedCount": 37, "failedCount": 1, "failures": [ { "name": "fraud-detection-scheduled-20260601020000-a1b2c3d4", "reason": "cluster unavailable" } ] } ``` CMF returns one of the following statuses: - `200 OK` when the request completed, including when zero candidates matched. In that case, `deletedCount` is `0` and `failures` is empty. - `400 Bad Request` when `source` is missing or invalid, or when both `olderThanDays` and `olderThanHours` are set. - `404 Not Found` when the environment or parent resource does not exist. - `409 Conflict` when every failure is due to the Flink cluster not being ready and `force` is not set. - `500 Internal Server Error` when every failure is due to some other error. The response body summarizes the failure reasons. ### Interoperability with the Confluent Platform for Apache Flink Kubernetes Operator The Confluent Platform for Apache Flink Kubernetes Operator (FKO) has its own periodic savepoint setting: `kubernetes.operator.periodic.savepoint.interval`. You can use either CMF scheduling or the FKO setting for a given resource, but not both: - CMF schedule only. CMF triggers and manages the savepoints, including retention. - FKO interval only. The FKO triggers the savepoints. CMF adopts them with `status.source` set to `SCHEDULE`, and the schedule `state` is reported as `FKO_MANAGED`. CMF retention does not apply to these savepoints; the FKO’s own retention configuration (for example, `kubernetes.operator.savepoint.history.max.count`) continues to apply. - Both set on the same resource. CMF rejects the request with HTTP 400. Configure one or the other. ### Automatic cleanup of failed periodic savepoints In addition to retention of completed savepoints, CMF automatically removes scheduled savepoints that ended in the `FAILED` or `ABANDONED` state after they are older than a retention window. Because these savepoints have no backing data in the blob store, only the CMF record is removed. Failed manual and upgrade savepoints are never cleaned up automatically. CMF administrators can tune the cleanup task with the following application properties (defaults shown): - `cmf.savepoint.periodic-cleanup-enabled: true`. Master switch for the cleanup task. When `false`, CMF runs no retention cleanup, no freshness SLA check, and no cleanup of failed scheduled savepoints, even if `cmf.savepoint.failed-periodic-cleanup-enabled` is `true`. - `cmf.savepoint.periodic-cleanup-interval-mins: 60`. How often, in minutes, the cleanup task runs. - `cmf.savepoint.failed-periodic-cleanup-enabled: true`. Enables automatic deletion of failed scheduled savepoints. - `cmf.savepoint.failed-periodic-retention-hours: 168`. How long, in hours, to keep failed scheduled savepoints before deleting them. The default is one week. ### Troubleshoot a schedule The schedule is registered but no savepoints appear. : Check `status.savepointSchedule.state`. If the state is `PAUSED`, `FKO_MANAGED`, or `DISABLED`, resolve that first (see the state table in [Monitor a schedule](#cmf-periodic-savepoints)).
If the state is `ACTIVE`, verify the parent resource is running: applications require `status.jobStatus.state` = `RUNNING`. Statements require `status.phase` = `RUNNING` and a compiled plan. Trigger attempts on a resource that has not compiled its plan yet are skipped. `SELECT` statements are never scheduled: CMF validates a `savepointSchedule` on a `SELECT` statement but does not register it, and `SELECT` statements do not inherit an environment default schedule.
Also check `nextTriggerTime`: if it is inside a blackout window, the trigger is skipped and the schedule advances to the next scheduled time. Scheduled savepoints stop after some time, and `state` shows `DISABLED`. : CMF trips a circuit breaker after five consecutive trigger failures. `lastErrorMessage` contains the last failure reason. Only failures to create the savepoint resource count. A savepoint that is created but then fails in Flink, for example because the savepoint directory is not accessible, does not trip the circuit breaker. Check `status.state` and `status.error` on the scheduled savepoints instead. Fix the underlying cause, such as by restoring connectivity to the Kubernetes cluster. Then edit any trigger-group field of the schedule to reset the failure counter and return the schedule to `ACTIVE`. The trigger-group fields are `cronExpression`, `timezone`, `paused`, `jitterSeconds`, `blackoutWindows`, and `catchUpOnRestart`. Editing only `retention` does not re-enable a disabled schedule. Triggers fire more than expected, or skip, at the same time of day. : A cron expression evaluated in a time zone that observes daylight saving time fires twice when clocks move back and once fewer when clocks move forward. If you need every interval to be exactly the same length, set `timezone: UTC`. Trigger attempts skip silently with no state change. : The trigger task skips (and advances `nextTriggerTime` without changing `lastTriggerTime`) when the resource is not running, when the current time is inside a blackout window, or when a previous scheduled savepoint for the same resource is still in flight. These skips are recorded as `INFO` lines in the CMF server log. ## Role permissions for savepoints Permissions for managing savepoints are handled with either attached savepoint permissions or detached savepoint permissions. ### Attached savepoints Permissions are inherited from the parent Flink application or statement: - Create or detach: Requires `EDIT` permission on the parent. You can only detach savepoints for Flink applications. - View: Requires `VIEW` permission on the parent. - Delete: Requires `REMOVE` permission on the parent. - Configure a savepoint schedule or pin a savepoint: Requires `EDIT` permission on the parent. - Bulk delete savepoints: Requires `REMOVE` permission on the parent. ### Detached savepoints Detached savepoints are governed by the `FlinkDetachedSavepoint` RBAC resource. For detailed information about role permissions for detached savepoints, see [Role Permissions for detached savepoints](../configure/access-control.md#cmf-access-control-detached-savepoints) in the access control documentation. For information about how to assign these roles and manage permissions, see [Configure Access Control for Confluent Manager for Apache Flink](../configure/access-control.md#cmf-access-control). ## Current limitations - Deleting attached savepoints: Deletion of an attached savepoint and its physical data can only be performed when the corresponding Flink job is running. - Deleting detached savepoints: Deleting a detached savepoint from CMF only removes the metadata. The underlying physical data in the blob store is not deleted and must be cleaned up manually. - Statement detach: Savepoints attached to a statement cannot be detached. - Statement restore: Statements cannot be started from a detached savepoint, and `startFromSavepoint` is accepted only when updating a statement. - Deleting parent resources: You cannot delete a Flink application or statement while it still has attached savepoints. You must delete (or detach, for applications) them first. - Schedule configuration surfaces: Savepoint schedules can only be configured through the CMF REST API. The Confluent CLI does not manage schedules, and applying a spec with `savepointSchedule` directly through `kubectl` against the FKO CRD silently drops the field. - Schedule scope: CMF scheduling and the FKO `kubernetes.operator.periodic.savepoint.interval` setting cannot be used on the same resource. See [Interoperability with the Confluent Platform for Apache Flink Kubernetes Operator](#cmf-periodic-savepoints-fko). - Retention scope: Automatic retention and cleanup only apply to savepoints with `status.source: SCHEDULE`. Manual and upgrade savepoints, and any pinned scheduled savepoint, are never deleted automatically. - Restore automation: CMF does not automatically restart a job from the latest scheduled savepoint. To restore, use `startFromSavepoint` as described in [Start a job from a savepoint](#cmf-savepoints). - Bulk delete scope: The bulk-delete endpoint is scoped to one application or statement per call and does not support environment-wide bulk delete. ## Related content - [REST APIs for Confluent Manager for Apache Flink](../clients-api/rest.md#af-rest-api) - [Deploy and Manage Confluent Manager for Apache Flink Applications](applications/overview.md#cmf-applications) - [Deploy and Manage Statements in Confluent Manager for Apache Flink](sql-statements/overview.md#flink-sql-statements) - [Application Instances in Confluent Manager for Apache Flink](applications/application-instances.md#cmf-application-instances) - [Describe, Update and Delete Confluent Manager for Apache Flink Applications](applications/manage.md#cmf-manage-apps)