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.

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.

The following steps use the CMF REST API to replace the statement orders-enrich-v1 in the environment env-1 with orders-enrich-v2.

Get the reference statement’s resource and save it to a file:

curl http://cmf:8080/cmf/api/v1/environments/env-1/statements/orders-enrich-v1 \
  -o orders-enrich-v1.json

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

Wait until the reference statement is in the STOPPED phase:

curl http://cmf:8080/cmf/api/v1/environments/env-1/statements/orders-enrich-v1

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:

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

Create the new statement:
```
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.
Check the new statement:
```
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.
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.
```
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 and Savepoint Management in Confluent Manager for Apache Flink.

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:

cmf:
  sql:
    carryOverOffsets:
      enabled: true

helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
  --version "~2.5.0" \
  --namespace <cmf_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.

Carry Over Offsets Between Statements in Confluent Manager for Apache Flink

Replace a statement

Limitations

Turn on carry-over offsets

Related content