Building a Dead Letter Queue with Kafka Connect and S3 Sink

When a Kafka Connect S3 Sink connector hits a record it cannot deserialize, transform, or write, the default behavior (errors.tolerance=none) is to fail the task — stalling every partition that task owns until a human intervenes. A single poison-pill message can halt an entire archival pipeline. The robust alternative is a Dead Letter Queue (DLQ): a Kafka topic into which Connect diverts failed records so the primary flow keeps moving. This guide builds that DLQ end to end, and is blunt about the part most write-ups get wrong: exactly which failures the built-in DLQ does and does not capture.

Understanding the Failure Modes in S3 Sink Pipelines

Before configuring anything, you need to know precisely where records can fail in a sink connector, because Kafka Connect’s built-in DLQ only catches some of those stages. The S3 Sink connector runs as a consumer within the broader Producers, Consumers & Delivery Semantics model, pulling records from its assigned partitions and writing them as objects. Records flow through three stages:

1. Conversion (deserialization). The key and value converters turn raw bytes from the topic into Connect records. This stage fails when a producer changes its serialization format, or when a rogue producer writes plain strings to a topic the converter expects to be Avro.
2. Transformation. Single Message Transforms (SMTs) run next. A Cast or field-extraction SMT throws if the record’s structure does not match its assumption.
3. Delivery (the connector’s put() method). The connector buffers records and uploads them to S3. Failures here include credential errors, missing buckets, and AWS API throttling.

The distinction that matters operationally: Kafka Connect’s built-in DLQ (KIP-298) only captures failures in the conversion and transformation stages. Errors raised inside the connector’s put() method — the actual S3 upload — are not automatically routed to the DLQ by the framework. A connector can opt into reporting bad records from put() to the same DLQ by using the ErrantRecordReporter interface added in KIP-610 (Apache Kafka 2.6), but that is connector-specific behavior, not a free feature of errors.tolerance.

This is the right design. Conversion and transform failures are deterministic and record-specific: the record itself is the problem, so a DLQ is appropriate. S3 upload failures are usually transient and infrastructure-level — credential expiry, a missing bucket, or throttling — and should be handled with retries and a halted task that pages someone, not silently shoveled into a DLQ. The same separation of permanent from transient failures appears in Implementing Dead Letter Queues in Kafka Consumers.

Mapping each failure to the mechanism that actually handles it keeps you from configuring something that will never fire:

Configuring the Kafka Connect Dead Letter Queue

Kafka Connect provides native DLQ support through a set of connector-level errors.* properties (DLQ output is supported for sink connectors only). The three that turn the feature on are:

• errors.tolerance — default none. Set to all to skip problematic records instead of failing the task. The DLQ is only written when this is all.
• errors.deadletterqueue.topic.name — default empty. The target topic. If empty, failed records are skipped (and optionally logged) but not preserved.
• errors.deadletterqueue.context.headers.enable — default false. Adds error-context headers to each DLQ record.

A minimal S3 Sink configuration with DLQ routing enabled:

{
  "name": "s3-sink-dlq",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "1",
    "topics": "events-raw",
    "s3.bucket.name": "my-production-bucket",
    "s3.region": "us-east-1",
    "flush.size": "1000",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.avro.AvroFormat",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "s3-sink-dlq",
    "errors.deadletterqueue.topic.replication.factor": "3",
    "errors.deadletterqueue.context.headers.enable": "true",
    "errors.log.enable": "true",
    "errors.retry.timeout": "300000",
    "errors.retry.delay.max.ms": "60000"
  }
}

Two retry properties govern how long Connect retries a failed operation before giving up on it:

• errors.retry.timeout — default 0 (no retries). The maximum total time, in milliseconds, that a failed operation is reattempted. Use -1 for infinite retries.
• errors.retry.delay.max.ms — default 60000. The maximum delay between consecutive retry attempts; Connect backs off with jitter up to this ceiling.

Note the exact property names: errors.retry.timeout and errors.retry.delay.max.ms. There is no errors.retry.delay.max without .ms, and errors.retry.timeout is not suffixed with .ms despite being a millisecond value.

One easy-to-miss default: errors.deadletterqueue.topic.replication.factor is 3. On a single-broker dev cluster the connector will fail to auto-create the DLQ topic unless you override this to 1.

Understand exactly what errors.tolerance=all does to your delivery guarantee. A skipped record’s offset is still committed — the connector moves on as if the record were delivered. For converter/transform failures with a DLQ topic configured, the record is preserved before that commit, so you have not lost it. But with errors.tolerance=all and no DLQ topic set, a failing record is dropped permanently and its offset committed: silent data loss by configuration. Treat the empty-topic.name case as a footgun, not a default. errors.tolerance is also all-or-nothing — there is no per-error-type or percentage threshold; once it is all, every tolerated stage skips on failure.

Securing the DLQ Topic on a Hardened Cluster

A subtlety that bites the first time you enable a DLQ on a secured cluster: writing to the DLQ uses two clients with separate code paths. Failed records are sent through a dedicated producer, and the DLQ topic itself is created by an admin client. On a plaintext dev cluster both inherit the worker’s broker connection and just work. On a SASL/SSL cluster, neither one automatically picks up the connector’s consumer credentials — so the connector starts, processes fine, and then fails the instant the first bad record needs a DLQ, with an authorization or timeout error that looks unrelated to error handling.

Configure both clients explicitly on the Connect worker. The worker derives the DLQ producer from producer.* properties and the topic-creating admin client from admin.* properties; you can also override per connector with producer.override.* and admin.override.*:

# Connect worker config (secured cluster)
admin.security.protocol=SASL_SSL
admin.sasl.mechanism=PLAIN
admin.ssl.endpoint.identification.algorithm=https
admin.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule \
  required username="connect" password="${secret}";

The service principal these clients use needs Create on the DLQ topic (or Describe/Write if you pre-create it) plus Write to produce. If you pre-create the topic, the admin client only needs to describe it, which is the safer least-privilege posture.

Designing the DLQ Topic and Retention Policy

Connect auto-creates the DLQ topic if it does not exist, but in production you should pre-create it so you control partition count, replication, and retention. Unlike application topics, a DLQ accumulates records that require human intervention; without a retention policy it grows unbounded.

kafka-topics --bootstrap-server localhost:9092 \
  --create \
  --topic s3-sink-dlq \
  --partitions 6 \
  --replication-factor 3 \
  --config retention.ms=604800000 \
  --config cleanup.policy=delete

This creates the DLQ with a 7-day (604800000 ms) retention window and time-based deletion. Use cleanup.policy=delete, not compact: failed records frequently share keys (or are tombstones), and log compaction would discard diagnostic records you still need. Size partition count to your throughput, not to tasks.max — Connect produces DLQ records keyed by the original record key, and a low-tasks.max connector can still drive a high failure rate during an incident.

Pick the retention window deliberately: it is your remediation deadline. If a record sits in the DLQ longer than retention.ms, the broker deletes it and the original payload is gone for good — so the window must comfortably exceed your worst-case time to detect, fix the root cause, and replay. Seven days suits a team with business-hours on-call; tighten it only if you are confident in your alerting.

Monitor DLQ depth and the lag of any consumer reading it. A sudden spike in DLQ writes almost always signals a systemic event — a Schema Registry outage or a producer that changed formats — rather than isolated bad records.

What the DLQ Record Looks Like for Deserialization Failures

Deserialization is the most common failure mode. When the value converter throws, Connect cannot produce a Connect record at all — so it writes the original raw bytes (key, value, and headers) of the failed message to the DLQ unchanged. It is not a tombstone and it is not re-encoded; downstream tooling must treat the DLQ value as opaque bytes.

The diagnostic information lives in the headers, which is why errors.deadletterqueue.context.headers.enable=true matters. With it enabled, Connect adds string-valued headers including:

• __connect.errors.topic, __connect.errors.partition, __connect.errors.offset — the source coordinates of the record.
• __connect.errors.stage and __connect.errors.class.name — which pipeline stage and class failed (for example, the converter class).
• __connect.errors.exception.class.name, __connect.errors.exception.message, __connect.errors.exception.stacktrace — the failure itself.
• __connect.errors.connector.name, __connect.errors.task.id — provenance.

Note the exact spelling: it is __connect.errors.exception.class.name (not ...exception.class), and there is no __connect.errors.timestamp header — the failure time is exposed as a JMX metric (last-error-timestamp), not a record header.

To see the failed payloads in the worker log as well, set errors.log.enable=true and errors.log.include.messages=true. The latter writes record contents to the Connect log, which is invaluable for debugging but dangerous with sensitive data — prefer the DLQ topic for inspection, or redact before logging. A typical converter block:

{
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://schema-registry:8081",
  "value.converter.enhanced.avro.schema.support": "true",
  "errors.tolerance": "all",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true"
}

enhanced.avro.schema.support is a real Confluent AvroConverter option (it preserves Avro enums and fixed types through the conversion); it does not change DLQ behavior. There is no converter setting that “passes through raw bytes on success” — raw-byte passthrough to the DLQ is exactly what already happens on a conversion failure.

Routing and Classifying DLQ Records

A single DLQ topic with rich error headers is sufficient for almost every team: a downstream consumer reads the topic, branches on __connect.errors.exception.class.name, and alerts or replays accordingly. Connect does not provide a built-in way to fan errors out to multiple DLQ topics by error type.

Be wary of advice to “classify errors with an SMT.” SMTs such as ReplaceField operate on a record’s own fields (its key or value Struct/Map) and run before a failure occurs — they cannot read __connect.errors.* headers, which only exist on records that have already been diverted to the DLQ. There is no transform chain that re-classifies records after they enter the DLQ. Classification belongs in the DLQ consumer, not in an SMT.

If you genuinely need per-error-type routing or want to forward put()-stage (S3 upload) failures into a DLQ, that requires connector code using the ErrantRecordReporter (KIP-610) inside the connector’s put() method — it is a connector implementation concern, not connector configuration. For the stock Confluent S3 Sink connector, treat the built-in DLQ as a converter/transform DLQ and handle delivery failures with retries and task-level alerting.

Operational Monitoring and Alerting for the DLQ

A DLQ without monitoring is a liability: failed records sitting unread are silent data loss. Kafka Connect exposes per-task error metrics via JMX under the task-error-metrics MBean, scoped by connector and task:

kafka.connect:type=task-error-metrics,connector=<name>,task=<id>

Useful attributes on this MBean:

• total-record-errors, total-record-failures, total-records-skipped — error and skip counts.
• total-retries, total-errors-logged — retry and logging activity.
• deadletterqueue-produce-requests, deadletterqueue-produce-failures — DLQ write attempts and failed DLQ writes.
• last-error-timestamp — epoch time of the most recent error.

There is no deadletterqueue-produce-bytes metric, and these attributes live under task-error-metrics, not under connect-worker-metrics. Watch deadletterqueue-produce-failures as carefully as the record counts: a nonzero value means failed records are not even reaching the DLQ — usually the security or replication-factor problems above — which is the one failure mode that turns a DLQ from a safety net into a hole. Scrape these with the Prometheus JMX exporter:

# jmx_exporter_config.yml
rules:
  - pattern: "kafka.connect<type=task-error-metrics, connector=(.+), task=(.+)><>deadletterqueue-produce-requests"
    name: "kafka_connect_dlq_produce_requests"
    labels:
      connector: "$1"
      task: "$2"
  - pattern: "kafka.connect<type=task-error-metrics, connector=(.+), task=(.+)><>total-record-failures"
    name: "kafka_connect_task_record_failures"
    labels:
      connector: "$1"
      task: "$2"

Complement task-level metrics with the lag of the consumer that drains the DLQ. Check it with kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group <dlq-consumer-group>, or from the broker-exported kafka_consumergroup_lag series if you use the Kafka Lag Exporter. Alert on sustained DLQ lag — for example, more than a few hundred un-consumed records for over 15 minutes — so on-call investigates before the retention window expires and the evidence is gone. See the Confluent Connect monitoring reference for the full metric list.

Reprocessing and Remediation Strategies

Once records land in the DLQ, the workflow shifts to remediation: either fix the root cause and replay, or accept the loss deliberately. Start with diagnosis — read the headers:

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic s3-sink-dlq \
  --from-beginning \
  --property print.headers=true \
  --property print.key=true \
  --max-messages 5

Because the DLQ value is the original raw bytes (and may be binary Avro), printing the value is usually noise; the __connect.errors.exception.class.name and __connect.errors.exception.message headers tell you what actually failed. Common offenders: org.apache.kafka.connect.errors.DataException for conversion/schema problems, and a Schema-Registry-related exception (HTTP errors against the registry) when the registry is unreachable or a subject is missing.

For reproducible failures, fix the cause — register the missing schema, correct the converter config, or adjust the SMT — then replay. Replay is not built into Connect: write a small consumer/producer that reads the DLQ and re-produces each record’s original key and value back to the source topic (the source coordinates are in the __connect.errors.topic/partition/offset headers). For large backlogs, parallelize with multiple consumers in a group.

Two replay hazards are worth pre-empting. First, the S3 Sink writes by topic/partition/offset, so re-producing a record to the source topic gives it a new offset — if your replayer is not idempotent and a record fails again, it lands back in the DLQ as a fresh entry, and naive loops can amplify. Cap replay attempts and track a redelivery count. Second, replay only the records whose root cause you have actually fixed; replaying the whole DLQ before the upstream producer or registry is corrected just refills it. Filter on __connect.errors.exception.class.name so you replay one failure class at a time.

Records that cannot be fixed — for instance, those emitted by a decommissioned service — should be archived and their offsets committed so the DLQ does not become a permanent graveyard. If you choose to retain them in object storage, point a second S3 Sink connector at the DLQ topic and apply an S3 lifecycle policy to transition the archived objects to cold storage.