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), serialization, or windowed aggregations are not routed to the DLQ. For recommendations on handling errors inside UDFs, see 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.

Prerequisites

Access to Confluent Cloud.
A provisioned Flink compute pool.
The appropriate RBAC roles and ACL permissions. For details, see Dead Letter Queue permissions.

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.

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.

Configure a DLQ for an existing table

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

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 the DLQ was created successfully, check that the DLQ topic and schema exist in your Kafka cluster and Schema Registry after running the statement.

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.

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:

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:

Create the Kafka topic with your desired configuration (partition count, retention, cleanup policy) using the Confluent Cloud Console, CLI, or API.
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 for the required schemas.

Configure your source table to use the pre-created DLQ:

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

DLQ table schema reference

The DLQ table uses the following schema. The error_timestamp field is stored in the Kafka message key, and all other fields are stored 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: `topic`, `partition`, `offset`, `timestamp`, `timestamp_type`, `headers`, `key` (bytes), and `value` (bytes).

Schema definitions for manual registration

When pre-creating a DLQ outside of 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

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

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

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