<a id="tableflow-configure"></a>

# Configure Tableflow in Confluent Cloud

Tableflow in Confluent Cloud provides table-level settings you can configure,
including snapshot retention for time-travel queries, data time-to-live
(TTL) for automatically expiring old rows, and an error-handling mode
for managing per-record materialization failures.

## Snapshot retention

Snapshot retention involves managing metadata that enables you to query a
previous state of your table, also known as “time-travel queries”. Tableflow
creates a snapshot every time it commits a change to your table. This includes
any time Tableflow adds or updates data to your table, and when it performs
maintenance tasks, like compaction.

Tableflow always maintains a minimum number of snapshots, but you can
configure how long additional snapshots should be retained before they are
expired by setting the `retention_ms` configuration. When a snapshot is past
its expiration time or more than 100 snapshots exist, Tableflow
removes the snapshot from the table asynchronously, as well as any data files
that are no longer needed by any remaining snapshots.

<a id="tableflow-configure-data-ttl"></a>

## Data TTL

With the data time-to-live (TTL) setting, you can configure Tableflow to
automatically delete old data in your tables. Tableflow deletes table rows
when they become older than the retention period that you specify.
By setting the TTL, you can reduce storage costs for situations where you only
need data for a limited time. For example, data TTL is useful if you analyze
only recent data or if you run periodic aggregations, such as hourly or daily
summaries.

#### NOTE
Data TTL in Tableflow is available to some Confluent customers as a
Limited Availability feature. For those with access, Confluent fully
supports data TTL and recommends it for production use. To request access,
contact [Confluent Support](https://support.confluent.io).

Tableflow measures the age of a row based on the timestamp of the
corresponding Kafka record.

Data TTL differs from snapshot retention, which manages metadata for time-travel
queries but does not delete rows from your table.

### How data TTL works

When you configure data TTL on a table, Tableflow asynchronously deletes each
row that exceeds your retention period.

First, Tableflow logically deletes expired rows in your table after a waiting
period. Tableflow waits up to 24 hours after at least one row in a Parquet
file expires. Then, Tableflow logically deletes the expired rows, which makes
them invisible to queries. If all rows in the file expire within that time,
Tableflow will delete the file. Whereas, if only some rows expire after 24
hours, Tableflow will remove those rows. This action requires *compaction*,
which means Tableflow creates a new file without the expired rows. The waiting
period minimizes two impacts on query performance: the logical deletions and the
compaction.

Subsequently, Tableflow physically deletes the Parquet file. Once every
snapshot that references a Parquet file expires, Tableflow physically deletes
the file from storage. By waiting for snapshots to expire, Tableflow ensures
that you can do time-travel queries for as long as your snapshot retention
settings permit.

### Configure data TTL

To configure the data TTL setting, use the Confluent Cloud API. Set the
`data_retention_ms` parameter when you create or update a table. Specify the
retention period in milliseconds. Tableflow rounds up this period to the
nearest day.

The following example configures data TTL to retain data for 90 days:

```json
{
  "spec": {
    "display_name": "high-volume-events",
    "config": {
      "data_retention_ms": "7776000000"
    },
    "table_formats": ["ICEBERG"],
    "environment": {"id": "env-xxxxx"},
    "kafka_cluster": {"id": "lkc-xxxxx"}
  }
}
```

<a id="tableflow-configure-error-handling-mode"></a>

## Error-handling mode

Tableflow offers three modes for handling per-record materialization failures:
`suspend`, `skip`, and `log`. The default mode is `suspend`. Assign the
value to the `error-handling.mode` table property, for example:

```sql
ALTER TABLE my_table SET (
 'error-handling.mode' = 'log'
);
```

You can configure error-handling mode by using Tableflow configurations or
the Flink [error-handling.mode](../../../flink/reference/statements/create-table.md#flink-sql-create-table-with-error-handling-mode) table property.

### Suspend mode

The default mode, `suspend`, causes Tableflow to enter the `Pause` state
whenever a record can’t be materialized and added to the table. This means that
in situations where your topic ingests a corrupted record, Tableflow will
pause processing on that record.

Failures that occur for reasons that are not record-specific always cause
Tableflow to enter the `Pause` state, regardless of the configured
`error-handling.mode`. This includes, but is not limited to, catalog- and
storage-access related errors and invalid schema changes.

### Skip mode

When the Tableflow failure strategy is set to `skip`, Tableflow skips
over records that fail to materialize. Tableflow reports the number of
skipped records in the `rows_skipped` metric.

### Log mode

When the Tableflow failure strategy is set to `log`, Tableflow sends
records that fail to materialize to a Dead Letter Queue (DLQ) table. The DLQ is
an Apache Kafka® topic in the same cluster as your Tableflow table. When records are
sent to the DLQ, they are also recorded in the `rows_skipped` metric.

#### IMPORTANT
Failure handling with log mode is available only for topics that are
configured with Avro or Protobuf schemas.

### Error-handling log target

When the Tableflow error-handling mode is set to `log`, you can optionally
set a specific topic by creating a new topic or using an existing topic for
your DLQ. The DLQ topic must follow the
[DLQ topic schema](#tableflow-configure-dlq-topic-schema).

If no destination for the DLQ is specified, Tableflow defaults to using a
topic named `error_log`.

The account enabling or updating Tableflow to use the `log` error-handling
mode must have write permissions on the topic that is used for the DLQ,
otherwise Tableflow encounters an error and enters the `Pause` state.

Assign the value to the `error-handling.log.target` table property, for
example:

```sql
ALTER TABLE my_table SET (
 'error-handling.log.target' = 'my_error_topic'
);
```

You can configure error-handling mode by using Tableflow configurations or
the Flink [error-handling.log.target](../../../flink/reference/statements/create-table.md#flink-sql-create-table-with-error-handling-log-target) table
property.

<a id="tableflow-configure-dlq-topic-schema"></a>

### DLQ topic schema

Tableflow DLQ topics must use the following Avro schema to operate correctly.
If you create the topic as part of DLQ enablement, it’s automatically registered
with this schema. For the full schema definitions in Avro, JSON Schema, and
Protobuf (including both key and value schemas), see
[Schema definitions for manual registration](../../../flink/how-to-guides/configure-dlq.md#flink-sql-configure-dlq-schema-definitions).

<details id="target-details">
<summary id="target-summary" style="display: list-item; cursor:pointer; color:#337ab7;">
  DLQ topic schema
</summary>
```json
{
  "fields": [
    {
      "name": "error_timestamp",
      "type": {
        "logicalType": "timestamp-millis",
        "type": "long"
      }
    },
    {
      "name": "error_code",
      "type": "int"
    },
    {
      "name": "error_reason",
      "type": "string"
    },
    {
      "name": "error_message",
      "type": "string"
    },
    {
      "name": "error_details",
      "type": {
        "type": "map",
        "values": [
          "null",
          "string"
        ]
      }
    },
    {
      "name": "processor",
      "type": "string"
    },
    {
      "default": null,
      "name": "statement_name",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "affected_type",
      "type": "string"
    },
    {
      "default": null,
      "name": "affected_catalog",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "default": null,
      "name": "affected_database",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "default": null,
      "name": "affected_name",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "default": null,
      "name": "source_record",
      "type": [
        "null",
        {
          "fields": [
            {
              "default": null,
              "doc": "The topic of the Kafka source record.",
              "name": "topic",
              "type": [
                "null",
                "string"
              ]
            },
            {
              "default": null,
              "doc": "The partition of the Kafka source record.",
              "name": "partition",
              "type": [
                "null",
                "int"
              ]
            },
            {
              "default": null,
              "doc": "The offset of the Kafka source record.",
              "name": "offset",
              "type": [
                "null",
                "long"
              ]
            },
            {
              "default": null,
              "doc": "The timestamp of the Kafka source record. The value is in milliseconds since epoch. The specific meaning of the timestamp depends on timestamp_type.",
              "name": "timestamp",
              "type": [
                "null",
                {
                  "logicalType": "timestamp-millis",
                  "type": "long"
                }
              ]
            },
            {
              "default": null,
              "doc": "The type of the timestamp in the Kafka source record. Possible values are 'CREATE_TIME' and 'LOG_APPEND_TIME'.",
              "name": "timestamp_type",
              "type": [
                "null",
                "string"
              ]
            },
            {
              "default": null,
              "doc": "The headers of the Kafka source record. The keys are strings and values are byte arrays. Note that for multiple headers with the same key, only the first one is kept in the map.",
              "name": "headers",
              "type": [
                "null",
                {
                  "type": "map",
                  "values": [
                    "null",
                    "bytes"
                  ]
                }
              ]
            },
            {
              "default": null,
              "doc": "The key of the Kafka source record. May be null if the record has no key.",
              "name": "key",
              "type": [
                "null",
                "bytes"
              ]
            },
            {
              "default": null,
              "doc": "The value of the Kafka source record. May be null if the record has no value (tombstone record).",
              "name": "value",
              "type": [
                "null",
                "bytes"
              ]
            }
          ],
          "name": "stocks_realtime_feed_dlq_value_source_record",
          "type": "record"
        }
      ]
    }
  ],
  "name": "stocks_realtime_feed_dlq_value",
  "namespace": "org.apache.flink.avro.generated.record",
  "type": "record"
}
```

</details>
<br/>

## Related content

- [Monitor Tableflow](monitor-tableflow.md#tableflow-monitor)
- [Storage with Tableflow](../concepts/tableflow-storage.md#tableflow-storage)
- [Grant Role-Based Access](tableflow-rbac.md#tableflow-rbac)

#### 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).