<a id="flink-sql-carry-over-offsets"></a>

# 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](../concepts/schema-statement-evolution.md#flink-sql-schema-and-statement-evolution)
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](../reference/statements/set.md#flink-sql-set-statement) statement, for example:

```sql
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](/cloud/current/api.html#tag/Statements-(sqlv1)/operation/createSqlv1Statement),
set the property in the `properties` map of the statement `spec`, for
example:

```json
{
  "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](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs/resources/confluent_flink_statement)
of the `confluent_flink_statement` resource, for example:

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

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

<a id="flink-sql-carry-over-offsets-statement-type-limitations"></a>

### 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](../concepts/statements.md#flink-sql-statements).
- 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](https://support.confluent.io/) (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](#flink-sql-carry-over-offsets-statement-type-limitations).

## Examples

### Statement already stopped

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

Create a new statement with updated logic:

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

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

## Related content

- [Schema and Statement Evolution](../concepts/schema-statement-evolution.md#flink-sql-schema-and-statement-evolution)
- [Flink SQL Shell Quick Start](../get-started/quick-start-shell.md#flink-sql-quick-start-shell)
- [Confluent CLI](../reference/flink-sql-cli.md#flink-sql-confluent-cli)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).