Carry-over Offsets in Confluent Cloud for Apache Flink

Carry-over offsets let a new Flink statement start from the topic offsets of an existing statement, without manually copying offsets between statements.

This lets you update Flink statements without data loss and reduces the need to monitor statement status when you deploy continuous integration and continuous delivery (CI/CD) pipelines.

Automatic orchestration handles the upgrade process. The system waits for the referenced statement to stop before starting the new statement, so processing transitions between statements without interruption.

You can use carry-over offsets only when you replace an existing statement. This feature lets you evolve statements with exactly-once semantics, but only when the statement is stateless. A stateless statement processes each event independently and in any order. The system determines automatically whether a statement is stateless.

This feature doesn’t apply to statements that use any of the following, because the update can cause inconsistent results:

Aggregates
LAG
Windows
Pattern matching
Upsert sinks

To use carry-over offsets, add the sql.tables.initial-offset-from property to your new statement’s configuration. In the Confluent Cloud Console and the Flink SQL shell, set the property by using the SET statement, for example:

SET 'sql.tables.initial-offset-from' = '<reference_statement_name>';

The <reference_statement_name> is the name of the statement to use as the reference for the carry-over offsets.

If you’re using the Statements API, set the property in the properties map of the statement spec, for example:

{
  "spec": {
    "properties": {
      "sql.tables.initial-offset-from": "<reference_statement_name>"
    }
  }
}

If you’re using the Confluent Terraform provider, set the property in the properties field of the confluent_flink_statement resource, for example:

resource "confluent_flink_statement" "new_statement" {
  properties = {
    "sql.tables.initial-offset-from" = "<reference_statement_name>"
  }
}

Considerations for carry-over offsets

Regional limitations

The referenced statement must be in the same organization, environment, and region as the new statement.
This property doesn’t support carry-over offsets across regions.

Timeout behavior

New statements wait up to six hours for the referenced statement to stop.
If the timeout expires, the new statement fails with an error message indicating the reason.

Table options priority

Explicit table options in your SQL text take precedence over carry-over offsets.
Only tables without explicit options use carry-over offsets.

Example of table options priority:

INSERT INTO output
  SELECT * FROM table1
UNION ALL
  SELECT * FROM table2 /*+ OPTIONS('scan.startup.mode' = 'latest-offset') */;

Result: table1 uses carry-over offsets, and table2 uses the specified latest-offset mode.

Statement type limitations

The referenced statement must be a background statement. You can’t use a foreground statement as the reference. For more information about these statement types, see Flink SQL Statements in Confluent Cloud for Apache Flink.
Stopping a background statement takes a savepoint, which is a consistent snapshot of the statement’s processing state. Carry-over offsets require this savepoint to determine the topic offsets for the new statement. Stopping a foreground statement doesn’t take a savepoint, so if you reference a foreground statement, the new statement has no offsets to carry over and remains in the Pending state indefinitely.

Common issues

Statement not found error

Verify the referenced statement name is correct.
Ensure the statement exists in the same organization, environment, and region.

Timeout exceeded

Check whether the referenced statement (the one named in sql.tables.initial-offset-from) is stopping.
Verify that no blocking conditions prevent termination.

Invalid SQL error

Confluent Cloud validates the new statement’s syntax as soon as you create it.
Fix SQL syntax errors before carry-over begins.

Referenced statement savepoint failed

The system couldn’t submit the statement because the referenced statement didn’t enter a stopped state gracefully. Data inconsistencies can occur when you use offsets from failed savepoints.
Try to resume the referenced statement and stop it again.
If the problem persists, contact Confluent Support (requires login).

New statement stuck in Pending state indefinitely

The new statement gets stuck in the Pending state indefinitely when the referenced statement is a foreground statement. Stopping a foreground statement doesn’t take a savepoint, so the new statement has no offsets to carry over and never leaves the Pending state.
Reference a background statement instead. For more information, see Statement type limitations.

Examples

Statement already stopped

Suppose you stopped a statement named my-original-statement.

Create a new statement with updated logic:

SET 'sql.tables.initial-offset-from' = 'my-original-statement';
INSERT INTO enhanced_output
SELECT
    user_id,
    event_type,
    `timestamp`,
    new_field
FROM user_events
WHERE event_type IN ('click', 'view', 'purchase');

Statement still running

Your original statement metrics-processor-v1 is still running. Create a new statement that references it:

SET 'sql.tables.initial-offset-from' = 'metrics-processor-v1';
INSERT INTO enhanced_output
SELECT
    user_id,
    event_type,
    `timestamp`,
    new_field
FROM user_events
WHERE event_type IN ('click', 'view', 'purchase');

The new statement remains in the Pending state until you stop metrics-processor-v1.