<a id="flink-sql-configure-dlq"></a>

# Configure a Dead Letter Queue

A Dead Letter Queue (DLQ) captures records that Confluent Cloud for Apache Flink cannot deserialize,
enabling your statements to continue processing instead of failing. When a
deserialization error occurs, Flink routes the problematic record to a DLQ table,
where you can inspect it later.

DLQ error handling applies only to deserialization errors at the source. Errors
in [user-defined functions (UDFs)](../concepts/user-defined-functions.md#flink-sql-udfs), serialization, or
windowed aggregations are not routed to the DLQ. For recommendations on handling
errors inside UDFs, see [Error handling best practices](create-udf.md#flink-sql-udf-error-handling-best-practices).

#### NOTE
Tableflow uses the same `error-handling.mode` and
`error-handling.log.target` table properties and the same DLQ schema. For
Tableflow-specific error handling behavior, see
[Error-handling mode](../../topics/tableflow/operate/configure-tableflow.md#tableflow-configure-error-handling-mode).

## Prerequisites

- Access to Confluent Cloud.
- A provisioned Flink compute pool.
- The appropriate RBAC roles and ACL permissions. For details, see
  [Dead Letter Queue permissions](../operate-and-deploy/flink-rbac.md#flink-rbac-dlq-permissions).

<a id="flink-sql-configure-dlq-new-table"></a>

## Configure a DLQ when creating a new table

You can enable DLQ error handling when creating a new table by setting the
`error-handling.mode` and `error-handling.log.target` table properties in
your `CREATE TABLE` statement.

```sql
CREATE TABLE my_source_table (
  id INT,
  name STRING,
  event_time TIMESTAMP_LTZ(3)
) WITH (
  'error-handling.mode' = 'log',
  'error-handling.log.target' = 'my_source_table_error_log'
);
```

Flink attempts to create the DLQ topic and register the schema automatically. If
you don’t specify `error-handling.log.target`, the default DLQ table name is
`error_log`.

<a id="flink-sql-configure-dlq-existing-table"></a>

## Configure a DLQ for an existing table

You can add DLQ error handling to an existing table by using an `ALTER TABLE`
statement.

```sql
ALTER TABLE my_source_table SET (
  'error-handling.mode' = 'log',
  'error-handling.log.target' = 'my_source_table_error_log'
);
```

Flink attempts to create the DLQ topic and register the schema at the time the
`ALTER TABLE` statement runs.

#### IMPORTANT
If DLQ creation fails (for example, the topic cannot be created, the schema
cannot be registered, or permissions are missing), the `CREATE TABLE` or
`ALTER TABLE` statement itself does not fail. The failure surfaces later
when the first deserialization error occurs and Flink cannot write the record
to the DLQ, causing the job to fail.

To verify that Flink created the DLQ successfully, check that the DLQ topic
and schema exist in your Kafka cluster and Schema Registry after running the statement.

<a id="flink-sql-configure-dlq-pre-create"></a>

## Pre-create a DLQ outside Flink

By default, Flink creates the DLQ topic and schema automatically with default
settings. If you need control over the topic configuration, such as partition
count, retention time, or cleanup policy, you can pre-create the DLQ topic and
schema before enabling error handling.

### Option A: Create the DLQ table in Flink with a `CREATE TABLE` statement

Run a `CREATE TABLE` statement with the full DLQ schema and your desired topic
configuration.

```sql
CREATE TABLE my_source_table_error_log (
  `error_timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL,
  `error_code` INT NOT NULL,
  `error_reason` VARCHAR(2147483647) NOT NULL,
  `error_message` VARCHAR(2147483647) NOT NULL,
  `error_details` MAP<VARCHAR(2147483647) NOT NULL, VARCHAR(2147483647)> NOT NULL,
  `processor` VARCHAR(2147483647) NOT NULL,
  `statement_name` VARCHAR(2147483647),
  `affected_type` VARCHAR(2147483647) NOT NULL,
  `affected_catalog` VARCHAR(2147483647),
  `affected_database` VARCHAR(2147483647),
  `affected_name` VARCHAR(2147483647),
  `source_record` ROW<
    `topic` VARCHAR(2147483647),
    `partition` INT,
    `offset` BIGINT,
    `timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE,
    `timestamp_type` VARCHAR(2147483647),
    `headers` MAP<VARCHAR(2147483647) NOT NULL, VARBINARY(2147483647)>,
    `key` VARBINARY(2147483647),
    `value` VARBINARY(2147483647)>
)
DISTRIBUTED INTO 6 BUCKETS
WITH (
  'changelog.mode' = 'append',
  'kafka.cleanup-policy' = 'delete',
  'kafka.retention.time' = '7 d',
  'value.format' = 'avro-registry'
);
```

Then configure your source table to use this DLQ table:

```sql
ALTER TABLE my_source_table SET (
  'error-handling.mode' = 'log',
  'error-handling.log.target' = 'my_source_table_error_log'
);
```

### Option B: Create the topic and schema manually

If you need full control and want to create the DLQ topic and schema outside of
Flink, follow these steps:

1. Create the Kafka topic with your desired configuration (partition count,
   retention, cleanup policy) using the Confluent Cloud Console, CLI, or API.
2. Register both a **key schema** and a **value schema** for the topic in Schema Registry.
   The subject names must follow the pattern `<topic_name>-key` and
   `<topic_name>-value`. See [Schema definitions for manual registration](#flink-sql-configure-dlq-schema-definitions)
   for the required schemas.
3. Configure your source table to use the pre-created DLQ:
   ```sql
   ALTER TABLE my_source_table SET (
     'error-handling.mode' = 'log',
     'error-handling.log.target' = 'my_source_table_error_log'
   );
   ```

<a id="flink-sql-configure-dlq-schema-reference"></a>

## DLQ table schema reference

The DLQ table uses the following schema. The `error_timestamp` field is
stored in the Kafka message key; all other fields appear in the message
value.

### Key fields

| Column            | Type               | Nullable   | Description              |
|-------------------|--------------------|------------|--------------------------|
| `error_timestamp` | `TIMESTAMP_LTZ(3)` | NOT NULL   | When the error occurred. |

### Value fields

| Column              | Type                  | Nullable   | Description                                                                                                                                                                                    |
|---------------------|-----------------------|------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `error_code`        | `INT`                 | NOT NULL   | Numeric error code.                                                                                                                                                                            |
| `error_reason`      | `STRING`              | NOT NULL   | Error category or reason.                                                                                                                                                                      |
| `error_message`     | `STRING`              | NOT NULL   | Human-readable error message.                                                                                                                                                                  |
| `error_details`     | `MAP<STRING, STRING>` | NOT NULL   | Additional key-value error details.                                                                                                                                                            |
| `processor`         | `STRING`              | NOT NULL   | The processor that encountered the error.                                                                                                                                                      |
| `statement_name`    | `STRING`              | Nullable   | Name of the statement that triggered the error.                                                                                                                                                |
| `affected_type`     | `STRING`              | NOT NULL   | Type of the affected resource.                                                                                                                                                                 |
| `affected_catalog`  | `STRING`              | Nullable   | Catalog of the affected table.                                                                                                                                                                 |
| `affected_database` | `STRING`              | Nullable   | Database of the affected table.                                                                                                                                                                |
| `affected_name`     | `STRING`              | Nullable   | Name of the affected table.                                                                                                                                                                    |
| `source_record`     | `ROW<...>`            | Nullable   | The original Kafka record that caused the error. Contains nested fields:<br/>`topic`, `partition`, `offset`, `timestamp`,<br/>`timestamp_type`, `headers`, `key` (bytes), and `value` (bytes). |

<a id="flink-sql-configure-dlq-schema-definitions"></a>

## Schema definitions for manual registration

When pre-creating a DLQ outside Flink, you must register both a **key schema**
and a **value schema** in Schema Registry. The subject names must follow the pattern
`<topic_name>-key` and `<topic_name>-value`.

#### NOTE
In the Avro schemas below, the `name` field uses `error_log` as a
placeholder (matching the default DLQ table name). Replace `error_log`
with your actual DLQ table name in the `name` and nested record `name`
fields.

### Key schema

### Avro

```json
{
  "fields": [
    {
      "name": "error_timestamp",
      "type": {
        "logicalType": "timestamp-millis",
        "type": "long"
      }
    }
  ],
  "name": "error_log_key",
  "namespace": "org.apache.flink.avro.generated.record",
  "type": "record"
}
```

### JSON Schema

```json
{
  "additionalProperties": false,
  "properties": {
    "error_timestamp": {
      "connect.index": 0,
      "connect.type": "int64",
      "title": "org.apache.kafka.connect.data.Timestamp",
      "type": "number"
    }
  },
  "required": ["error_timestamp"],
  "title": "Record",
  "type": "object"
}
```

### Protobuf

```protobuf
syntax = "proto3";
package org.apache.flink.proto.generated.record;

import "google/protobuf/timestamp.proto";

message Record {
  .google.protobuf.Timestamp error_timestamp = 1 [(confluent.field_meta) = {
    params: [
      { key: "flink.version", value: "1" },
      { key: "flink.notNull", value: "true" },
      { key: "flink.precision", value: "3" }
    ]
  }];
}
```

### Value schema

### Avro

```json
{
  "fields": [
    {
      "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,
              "name": "topic",
              "type": ["null", "string"]
            },
            {
              "default": null,
              "name": "partition",
              "type": ["null", "int"]
            },
            {
              "default": null,
              "name": "offset",
              "type": ["null", "long"]
            },
            {
              "default": null,
              "name": "timestamp",
              "type": [
                "null",
                {
                  "logicalType": "timestamp-millis",
                  "type": "long"
                }
              ]
            },
            {
              "default": null,
              "name": "timestamp_type",
              "type": ["null", "string"]
            },
            {
              "default": null,
              "name": "headers",
              "type": [
                "null",
                {
                  "type": "map",
                  "values": ["null", "bytes"]
                }
              ]
            },
            {
              "default": null,
              "name": "key",
              "type": ["null", "bytes"]
            },
            {
              "default": null,
              "name": "value",
              "type": ["null", "bytes"]
            }
          ],
          "name": "error_log_value_source_record",
          "type": "record"
        }
      ]
    }
  ],
  "name": "error_log_value",
  "namespace": "org.apache.flink.avro.generated.record",
  "type": "record"
}
```

### JSON Schema

```json
{
  "additionalProperties": false,
  "properties": {
    "error_code": {
      "connect.index": 0,
      "connect.type": "int32",
      "type": "number"
    },
    "error_reason": {
      "connect.index": 1,
      "type": "string"
    },
    "error_message": {
      "connect.index": 2,
      "type": "string"
    },
    "error_details": {
      "additionalProperties": {
        "oneOf": [
          { "type": "null" },
          { "type": "string" }
        ]
      },
      "connect.index": 3,
      "connect.type": "map",
      "type": "object"
    },
    "processor": {
      "connect.index": 4,
      "type": "string"
    },
    "statement_name": {
      "connect.index": 5,
      "oneOf": [
        { "type": "null" },
        { "type": "string" }
      ]
    },
    "affected_type": {
      "connect.index": 6,
      "type": "string"
    },
    "affected_catalog": {
      "connect.index": 7,
      "oneOf": [
        { "type": "null" },
        { "type": "string" }
      ]
    },
    "affected_database": {
      "connect.index": 8,
      "oneOf": [
        { "type": "null" },
        { "type": "string" }
      ]
    },
    "affected_name": {
      "connect.index": 9,
      "oneOf": [
        { "type": "null" },
        { "type": "string" }
      ]
    },
    "source_record": {
      "connect.index": 10,
      "oneOf": [
        { "type": "null" },
        {
          "additionalProperties": false,
          "properties": {
            "topic": {
              "connect.index": 0,
              "oneOf": [
                { "type": "null" },
                { "type": "string" }
              ]
            },
            "partition": {
              "connect.index": 1,
              "oneOf": [
                { "type": "null" },
                { "connect.type": "int32", "type": "number" }
              ]
            },
            "offset": {
              "connect.index": 2,
              "oneOf": [
                { "type": "null" },
                { "connect.type": "int64", "type": "number" }
              ]
            },
            "timestamp": {
              "connect.index": 3,
              "oneOf": [
                { "type": "null" },
                {
                  "connect.type": "int64",
                  "title": "org.apache.kafka.connect.data.Timestamp",
                  "type": "number"
                }
              ]
            },
            "timestamp_type": {
              "connect.index": 4,
              "oneOf": [
                { "type": "null" },
                { "type": "string" }
              ]
            },
            "headers": {
              "connect.index": 5,
              "oneOf": [
                { "type": "null" },
                {
                  "additionalProperties": {
                    "oneOf": [
                      { "type": "null" },
                      { "connect.type": "bytes", "type": "string" }
                    ]
                  },
                  "connect.type": "map",
                  "type": "object"
                }
              ]
            },
            "key": {
              "connect.index": 6,
              "oneOf": [
                { "type": "null" },
                { "connect.type": "bytes", "type": "string" }
              ]
            },
            "value": {
              "connect.index": 7,
              "oneOf": [
                { "type": "null" },
                { "connect.type": "bytes", "type": "string" }
              ]
            }
          },
          "title": "Record_source_record",
          "type": "object"
        }
      ]
    }
  },
  "required": [
    "error_code",
    "error_reason",
    "error_message",
    "error_details",
    "processor",
    "affected_type"
  ],
  "title": "Record",
  "type": "object"
}
```

### Protobuf

```protobuf
syntax = "proto3";
package org.apache.flink.proto.generated.record;

import "google/protobuf/timestamp.proto";

message Record {
  int32 error_code = 1;
  string error_reason = 2;
  string error_message = 3;
  repeated ErrorDetailsEntry error_details = 4;
  string processor = 5;
  optional string statement_name = 6;
  string affected_type = 7;
  optional string affected_catalog = 8;
  optional string affected_database = 9;
  optional string affected_name = 10;
  optional source_record_Row source_record = 11;

  message ErrorDetailsEntry {
    string key = 1;
    optional string value = 2;
  }

  message source_record_Row {
    optional string topic = 1;
    optional int32 partition = 2;
    optional int64 offset = 3;
    optional .google.protobuf.Timestamp timestamp = 4 [(confluent.field_meta) = {
      params: [
        { key: "flink.precision", value: "3" },
        { key: "flink.version", value: "1" }
      ]
    }];
    optional string timestamp_type = 5;
    headersRepeatedWrapper headers = 6 [(confluent.field_meta) = {
      params: [
        { key: "flink.wrapped", value: "true" },
        { key: "flink.version", value: "1" }
      ]
    }];
    optional bytes key = 7;
    optional bytes value = 8;

    message headersRepeatedWrapper {
      repeated ValueEntry value = 1;

      message ValueEntry {
        string key = 1;
        optional bytes value = 2;
      }
    }
  }
}
```

<a id="flink-sql-configure-dlq-inspect"></a>

## Inspect DLQ records

Query the DLQ table to view captured deserialization errors:

```sql
SELECT * FROM my_source_table_error_log;
```

Each record contains the error details and, when available, the original Kafka
record (topic, partition, offset, key, and value) that caused the error.

<a id="flink-sql-configure-dlq-monitor"></a>

## Monitor source deserialization errors

Confluent Cloud for Apache Flink exposes a per-statement counter of source deserialization failures as
the metric `io.confluent.flink/num_records_in_errors`.
The counter increments once per record that fails source deserialization,
independent of `error-handling.mode` — the mode only controls what happens
to the failed record itself:

- `log` (DLQ mode): the record is also written to the DLQ target table.
- `ignore`: the record is dropped silently.
- `fail`: the statement fails after the counter advances.

Use this metric to alert on the rate of source deserialization errors against
your own baseline, regardless of which error-handling mode the source table
uses.

The Confluent Cloud Metrics API exposes this counter alongside other Flink metrics.
Use the `table_name` label to break the count down per source table. For
details on querying or exporting it to third-party services such as Prometheus,
Grafana, or Datadog, see [Flink metrics integrations](../operate-and-deploy/monitor-statements.md#flink-sql-monitor-metrics). For the full list of
Flink metrics, see the
[Metrics API Reference](https://api.telemetry.confluent.cloud/docs/descriptors/datasets/cloud).

This counter increments only for source deserialization errors; errors in UDFs,
serialization, or windowed aggregations are not counted (the same limitation
that applies to the DLQ target table itself).

### Suggested alert

Alert when the share of failed records over a recent window exceeds a small
fraction of total ingested records. The threshold should be tuned to your own
baseline; for example, the following PromQL expression fires when more than
5% of ingested records have failed deserialization over a 15-minute window:

```text
increase(confluent_flink_num_records_in_errors[15m])
  / increase(confluent_flink_num_records_in[15m]) > 0.05
```

Compare against `io.confluent.flink/num_records_in` (the total input record
count for the same source table) so the alert tracks relative degradation
rather than the absolute error count, which naturally scales with throughput.
The same expression works in Grafana and Datadog with the corresponding
provider-specific metric names exported from the Metrics API.

<a id="flink-sql-configure-dlq-limitations"></a>

## Limitations and known behaviors

- DLQ error handling applies only to **deserialization errors** at the source.
  Errors in UDFs, serialization, or windowed aggregations are not routed to the
  DLQ. For UDF error handling, see
  [Error handling best practices](create-udf.md#flink-sql-udf-error-handling-best-practices).
- You configure error handling at the **table level**, not per statement.
  All statements reading from a table with `error-handling.mode` set to
  `log` share the same DLQ configuration.
- Each source table can have **one DLQ target** at a time. You set DLQ
  configuration at the source table level, not per job or per consumer.

<a id="flink-sql-configure-dlq-best-practices"></a>

## Best practices

- **Use a dedicated service account for DLQ setup** when consuming from
  shared or team-owned source topics. Because configuring a DLQ requires
  `DeveloperManage` permissions on the source topic, consider using a
  platform or admin service account to run the `ALTER TABLE` statement.
  This avoids granting elevated permissions on source topics to individual
  consumer service accounts.

## Related content

- [Data Contracts](../../sr/fundamentals/data-contracts.md#sr-data-contracts) (for Schema Registry
  rule-based DLQ routing)
- [User-Defined Functions](../concepts/user-defined-functions.md#flink-sql-udfs)
- [UDF Error Handling](create-udf.md#flink-sql-udf-error-handling-best-practices)
- [Data Type Mappings](../reference/serialization.md#flink-sql-serialization)
- [Schema Registry](../../sr/schemas-manage.md#sr-prv)
- [CREATE TABLE](../reference/statements/create-table.md#flink-sql-create-table)
- [ALTER TABLE](../reference/statements/alter-table.md#flink-sql-alter-table)
- [Flink RBAC](../operate-and-deploy/flink-rbac.md#flink-rbac)
- [Flink metrics integrations](../operate-and-deploy/monitor-statements.md#flink-sql-monitor-metrics)

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