# Carry Over Offsets Between Statements in Confluent Manager for Apache Flink Confluent Manager for Apache Flink® (CMF) can start a new SQL statement from the Apache Kafka® offsets that a stopped statement reached. Use this when you replace a statement with a new version, so that the new statement reads its topics where the old one stopped. #### IMPORTANT Only Kafka offsets carry over, not state. Use carry-over offsets only for stateless, append-only statements, such as statements that filter, project, or route records. CMF doesn’t check this. Before you begin, turn on carry-over offsets, which are off by default. For the Helm values, see [Turn on carry-over offsets](#cmf-carry-over-offsets-enable). ## Replace a statement You name the stopped statement in the new statement’s `sql.tables.initial-offset-from` property. CMF reads the offsets from the savepoint that Apache Flink® took when that statement stopped, and starts the new statement from them. The reference statement must be an `INSERT INTO` or `STATEMENT SET` statement in the same environment, with the `savepoint` upgrade mode, so that stopping it takes a savepoint. If the statement doesn’t set `spec.upgradeMode`, CMF uses `savepoint` only when the compute pool, directly or through the environment’s `computePoolDefaults`, sets both `state.checkpoints.dir` and `state.savepoints.dir`. Otherwise CMF uses `stateless`, and CMF rejects a statement that names the reference. For more information, see “Stop statements” in [Manage Statements in Confluent Manager for Apache Flink](manage-statements.md#flink-sql-statements-manage). The following steps use the CMF REST API to replace the statement `orders-enrich-v1` in the environment `env-1` with `orders-enrich-v2`. 1. Get the reference statement’s resource and save it to a file: ```bash curl http://cmf:8080/cmf/api/v1/environments/env-1/statements/orders-enrich-v1 \ -o orders-enrich-v1.json ``` 2. In `orders-enrich-v1.json`, set `spec.stopped` to `true` and remove `metadata.updateTimestamp`. CMF rejects an update that sets it, with HTTP `422`. Then stop the reference statement with a PUT request: ```bash curl -v -H "Content-Type: application/json" \ -X PUT http://cmf:8080/cmf/api/v1/environments/env-1/statements/orders-enrich-v1 \ -d @orders-enrich-v1.json ``` You can also stop it with `confluent flink statement stop`. For more information, see [Manage Statements in Confluent Manager for Apache Flink](manage-statements.md#flink-sql-statements-manage). 3. Wait until the reference statement is in the `STOPPED` phase: ```bash curl http://cmf:8080/cmf/api/v1/environments/env-1/statements/orders-enrich-v1 ``` 4. Create a file named `orders-enrich-v2.json` for the new statement. Set `sql.tables.initial-offset-from` in `spec.properties` to the name of the reference statement: ```json { "apiVersion": "cmf.confluent.io/v1", "kind": "Statement", "metadata": { "name": "orders-enrich-v2" }, "spec": { "statement": "INSERT INTO orders_enriched SELECT order_id, customer_id, price * quantity AS total FROM orders;", "properties": { "sql.current-catalog": "kafka-cat", "sql.current-database": "kafka-db", "sql.tables.initial-offset-from": "orders-enrich-v1" }, "computePoolName": "pool" } } ``` 5. Create the new statement: ```bash curl -v -H "Content-Type: application/json" \ -X POST http://cmf:8080/cmf/api/v1/environments/env-1/statements \ -d @orders-enrich-v2.json ``` If CMF returns HTTP `409`, read the message in the response. If it says that the reference statement’s savepoint is still being recorded, wait a few seconds and try again. If the savepoint can’t be used, for example because it failed, CMF returns HTTP `400` with a message that names the savepoint and the reason. To check the savepoints of the reference statement, see [Savepoint Management in Confluent Manager for Apache Flink](../savepoints.md#cmf-savepoints). 6. Check the new statement: ```bash curl http://cmf:8080/cmf/api/v1/environments/env-1/statements/orders-enrich-v2 ``` The new statement stays in the `PENDING` phase while CMF reads the offsets, and then moves to `RUNNING`. `status.detail` says what the statement waits for and the deadline after which it fails if the offsets aren’t available. While the statement waits, you can only change `spec.stopped`. Deleting the waiting statement cancels the read. After the statement is deployed, `status.traits.inheritedOffsetTopics` lists the topics that start from the inherited offsets. The list is empty if every table sets its own start with an `OPTIONS` hint. 7. After the new statement is running, delete the reference statement if you no longer need it. Delete its savepoints first. CMF doesn’t delete a statement that has savepoints attached, and returns HTTP `409`. Set `force=true`, because the reference statement isn’t running. Without it, CMF returns HTTP `409` for the savepoints too. ```bash curl -X DELETE \ "http://cmf:8080/cmf/api/v1/environments/env-1/statements/orders-enrich-v1/savepoints?source=UPGRADE&force=true" curl -X DELETE http://cmf:8080/cmf/api/v1/environments/env-1/statements/orders-enrich-v1 ``` If the reference statement also has manual or scheduled savepoints, delete them with `source=MANUAL` and `source=SCHEDULE`. Pinned savepoints aren’t deleted this way, so unpin them first. With `force=true`, CMF removes the savepoints but can leave their files in the savepoint directory. Delete the files there if you no longer need them. For more information, see [Filter and bulk-delete savepoints by source](../savepoints.md#cmf-periodic-savepoints-source) and [Savepoint Management in Confluent Manager for Apache Flink](../savepoints.md#cmf-savepoints). After the offsets are read, a GET request on the reference statement shows them in `status.latestOffsets`, as a map of topic name to offsets in the format of the `scan.startup.specific-offsets` table option. To create the statement with a `FlinkStatement` resource instead, set the property in `spec.properties`. The `confluent flink statement create` command can’t set the property. In the Confluent CLI shell, run `SET 'sql.tables.initial-offset-from' = 'orders-enrich-v1';`. The inherited offsets replace a `scan.startup.mode` that is set in the table’s `WITH` clause. An `OPTIONS` hint that sets `scan.startup.mode` on a table reference keeps that start for that reference. If a table reference has more than one `OPTIONS` hint, only the first one counts. ## Limitations - State doesn’t carry over. Statements with aggregations, joins, or windows start with empty state and can produce wrong results. - If Kafka deletes records because of retention before the new statement reads them, those records are lost. The new statement starts from the earliest offset that is still in the topic. - Only tables with the `kafka` or `confluent` connector and a single topic inherit offsets. `upsert-kafka` tables can’t inherit offsets. The statement fails if it reads no `kafka` or `confluent` table, or if a table with a topic list or `topic-pattern` reads an inherited topic. - The reference statement must already be stopped. CMF doesn’t wait for it to stop. ## Turn on carry-over offsets Add the following values to your CMF Helm values file, and then upgrade CMF: ```yaml cmf: sql: carryOverOffsets: enabled: true ``` ```bash helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \ --version "~2.5.0" \ --namespace \ --values cmf-values.yaml ``` | Helm value | Default | Description | |----------------------------------------|---------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `cmf.sql.carryOverOffsets.enabled` | `false` | Lets statements inherit offsets with `sql.tables.initial-offset-from`. | | `cmf.sql.carryOverOffsets.readerImage` | The Flink 2.3 `confluentinc/cp-flink-sql` image of your CMF release | The image that reads the savepoint. CMF uses this image regardless of the Flink version
of the compute pool. It also reads the savepoints of Flink 1.19 statements. Set it
only to pull the image from another registry. | | `cmf.sql.carryOverOffsets.timeout` | `15m` | How long a new statement waits for its offsets before it fails, counted from when
the statement is created. | If you turn carry-over offsets off, the statements that still wait for offsets fail. To read the offsets, CMF runs a short-lived Kubernetes `Job` in the environment’s namespace. The Job uses the compute pool’s JobManager pod template and runs as the environment’s service account. Its name starts with `cmf-coo-offsets-`. If the read fails, the new statement fails. Its `status.detail` shows the error, or names the Job whose logs to read. CMF needs permission to create, get, and delete `jobs` in the `batch` API group in the environment’s namespace. The Helm chart grants it when `rbac` is `true`, the default. If you set `rbac` to `false`, grant it to CMF yourself. To create a statement that inherits offsets, a user needs permission to view the reference statement. For more information, see [Configure Access Control for Confluent Manager for Apache Flink](../../configure/access-control.md#cmf-access-control). ## Related content - [Manage Statements in Confluent Manager for Apache Flink](manage-statements.md#flink-sql-statements-manage) - [Savepoint Management in Confluent Manager for Apache Flink](../savepoints.md#cmf-savepoints)