Error Handling & DLQ #

In large-scale distributed systems streaming millions of events every second, errors aren’t a question of “whether they’ll happen” but “when they’ll happen”. Inside Kafka Connect data pipelines, errors can come from various sources, from temporary network outages, authentication failures, to corrupted messages whose formats don’t match the schema (poison pills). If we don’t design a mature error handling strategy, a single corrupted message sent by a producer can crash the entire Connect worker cluster and stop our important business flow. To anticipate this, Kafka Connect provides built-in Error Handling features and the Dead Letter Queue (DLQ). This article will dissect the classification of failure types in Kafka Connect, explain fault tolerance settings, describe DLQ configuration, and discuss reliable automatic retry tactics in production.

Types of Failures in Kafka Connect (Classification of Failures) #

To handle errors precisely, we must first classify the failure types occurring inside Kafka Connect pipelines into three main categories:

1. Transient Network Failures #

These failures are temporary and usually caused by network infrastructure constraints or busy external systems.

  • Examples: Database connection timeouts, target systems restarting, or temporary network partitions between Connect workers and Kafka brokers.
  • Solution: This problem can be automatically resolved by retrying delivery after a certain pause.

2. Data Format Errors (Poison Pills / Serialization Errors) #

These errors are permanent for specific data records. Those records can never be decoded because they’re structurally corrupted.

  • Examples: Consumers expect Avro binary format, but producers send random text strings, or sent JSON is missing mandatory fields (non-nullable constraints).
  • Solution: Retrying the same data won’t solve the problem. These corrupted messages must be ignored or separated into a quarantine container (DLQ) so the pipeline can keep processing the next valid data.

3. Target System Logic Errors #

Errors thrown by external storage systems during the write process.

  • Examples: Running SQL INSERT operations violating unique constraint rules on target database tables, or sending documents with field structures conflicting with Elasticsearch indexes.
  • Solution: Requires special handling on the connector side (like setting operations to upsert mode to ignore key duplication).

Fault Tolerance Strategies (errors.tolerance) #

Kafka Connect provides an important configuration parameter called errors.tolerance to determine worker behavior when detecting corrupted data records or processing failures in the Converter, SMT, or Connector write stages.

There are two value options we can set on this property:

flowchart TD
    R1["Corrupted Data Record"] -->|errors.tolerance = none| T1["Task CRASH / STOP (Fail-Fast)"]
    R2["Corrupted Data Record"] -->|errors.tolerance = all| T2["Log Error / Send to DLQ (Tolerant)"]

1. errors.tolerance = none (Default / Fail-Fast) #

This is Kafka Connect’s built-in setting. If even the smallest error happens on one data record, the Connect task immediately throws an Exception, stops its thread execution, and changes its status to FAILED.

  • When to Use: Highly recommended for financial transaction data pipelines, billing, or other critical systems demanding 100% data consistency. It’s better for the system to stop completely than write wrong data or experience data loss (zero data loss guarantee).

2. errors.tolerance = all (Tolerant Mode) #

In this mode, if a data record processing error happens, the Connect worker logs that error, ignores the corrupted record, and immediately continues processing the next record without stopping the task.

  • When to Use: Suitable for telemetry log analysis systems, IoT metrics, or clickstream web activity tracking where losing a few data rows doesn’t significantly affect analytical results.
  • Risk: Enabling this property without a DLQ causes corrupted data to be lost forever without the developer team noticing.

What About Source Connectors? (Handling Source Errors) #

As discussed, Kafka Connect’s built-in Dead Letter Queue only works for Sink Connectors. So, how do we manage corrupted data records on the Source Connector side before they’re written to Kafka?

If a Source Connector encounters a corrupted database data row (for example binary text data that can’t be parsed by the converter):

  • Custom SMT Routing Pattern: We can write a custom Single Message Transform (SMT) plugin. This SMT is wrapped in a try-catch block. If payload parsing fails, the SMT changes the record’s destination topic name to a special topic (e.g., corrupted-database-records) instead of throwing an Exception that stops the task.
  • Filter Drop: If corrupted data at the source may be ignored, the custom SMT can filter that data by returning null, so the Connect worker doesn’t send that data to the Kafka broker.

Dead Letter Queue (DLQ) Configuration for Corrupted Records #

To balance between data pipeline continuity (so tasks don’t crash) and data safety (so corrupted data isn’t lost), we must combine "errors.tolerance": "all" with a Dead Letter Queue (DLQ).

A Dead Letter Queue is a special Kafka topic used to hold every data record that failed to be processed by a Sink Connector.

flowchart TD
    subgraph ConnectProcess["Connect Task Processing Loop"]
        Record["Record from Kafka Topic"] -->|1. Read| Converter{"Converter / SMT"}
        Converter -->|Success| Process["Send to Target System"]
        Converter -->|Failed / Corrupted| ErrorCheck{"errors.tolerance?"}
        
        ErrorCheck -->|none| Crash["Task Crash & Stop"]
        ErrorCheck -->|all| DLQCheck{"DLQ Configured?"}
        
        DLQCheck -->|Yes| WriteDLQ["Write to DLQ Topic + Headers"]
        DLQCheck -->|No| Skip["Ignore & Discard Record"]
        
        WriteDLQ & Skip & Process -->|Continue Loop| Record
    end

Storing Error Metadata in Message Headers #

When Kafka Connect routes a corrupted message to the DLQ topic, that message isn’t just sent as-is. We can configure workers to embed very valuable error metadata into the Kafka Connect message Headers.

The embedded metadata includes:

  • deadletterqueue.topic: The original topic name of the corrupted message.
  • deadletterqueue.partition: The original partition ID.
  • deadletterqueue.offset: The original offset position of the problematic message.
  • deadletterqueue.reason: The exception log message explaining why this record failed processing.
  • deadletterqueue.error.class: The name of the thrown Java Exception class.

With this metadata header, developer teams can easily build monitor applications that read the DLQ topic, analyze corruption causes, fix data, and automatically re-drive it back to the main pipeline.

Custom Error Logging & Alerting Mechanisms #

Besides filtering corrupted data to the DLQ, reliable production systems must have full visibility into when and why errors happen.

1. Log4j Configuration on Connect Workers #

We can control how detailed error logs are recorded into system log files. Inside the worker log4j configuration file (/opt/kafka/config/connect-log4j.properties), we must configure special error handling packages:

# ✓ CORRECT: Enabling debug detail to trace Connect error stacktraces
log4j.logger.org.apache.kafka.connect.runtime.WorkerSinkTask=DEBUG
log4j.logger.org.apache.kafka.connect.runtime.errors=DEBUG

These logs are then sent to centralized log systems (like Elasticsearch, Splunk, or Datadog) to trigger real-time alerting to Slack or PagerDuty when Exception counts surge above normal thresholds.

2. Monitoring via JMX Error Metrics #

Kafka Connect exposes specific error handling metrics that we must integrate with Prometheus/Grafana dashboards:

  • total-record-errors: The number of records that failed processing (experienced errors) since the task started.
  • total-record-failures: The number of records that triggered task failures (causing FAILED task status).
  • total-records-skipped: The number of corrupted records ignored/skipped because of the errors.tolerance=all setting.
  • deadletterqueue-produce-failures: The number of failures when workers try writing corrupted messages to the DLQ topic (for example from problematic ACL authorization).

Dead Letter Queue Re-drive Patterns #

Corrupted messages landing in the DLQ must not be left piling up forever without handling. We must design a Re-drive pattern (the process of reprocessing DLQ data):

flowchart TD
    DLQ["DLQ Kafka Topic"] --> App["Read via Re-drive App"]
    App --> B1["Network / Target Downtime?"] -- "Fix Target" --> WD["Write Direct"]
    App --> B2["Corrupted Data Format?"] -- "Fix Payload" --> RI["Re-ingest"]

1. Network Failure Scenario #

If data enters the DLQ because the target database downtime exceeded retry time limits:

  • The DevOps team fixes the target database until it’s back up normally.
  • A simple re-drive application reads messages from the DLQ and writes them directly to the target database using a standard database client.

2. Corrupted Payload Format Scenario #

If data enters the DLQ because the payload format is corrupted (for example the created_at column has an invalid data type):

  • We create a small microservice (or Kafka Streams application) subscribing to the DLQ topic.
  • That application reads messages, fixes their data format (for example parsing wrong date strings into standard ISO format), then republishes the cleaned messages back to the main business Kafka topic so the Sink Connector consumes them again.

Failed Task Restart Automation (Auto-Recovery Script) #

In dynamic production environments, Connect tasks can fail from transient network errors exceeding retry time limits. Instead of manually restarting via REST API commands every time an alert arrives, we can deploy a simple auto-recovery automation script running as a cron job in the Connect cluster:

#!/bin/bash
# Automation script to detect and restart failed Kafka Connect tasks

CONNECT_URL="http://localhost:8083"

# Fetch the list of all active connectors
CONNECTORS=$(curl -s "${CONNECT_URL}/connectors")

for connector in $(echo "${CONNECTORS}" | jq -r '.[]'); do
  # Check the health status of the connector and its tasks
  STATUS=$(curl -s "${CONNECT_URL}/connectors/${connector}/status")
  
  # Iterate over each task
  for task in $(echo "${STATUS}" | jq -r '.tasks[].id'); do
    STATE=$(echo "${STATUS}" | jq -r ".tasks[${task}].state")
    
    if [ "${STATE}" == "FAILED" ]; then
      echo "✗ Found task ${task} on connector ${connector} with FAILED status!"
      echo "Attempting to automatically restart the task..."
      
      # Send the restart command via the REST API
      curl -s -X POST "${CONNECT_URL}/connectors/${connector}/tasks/${task}/restart"
      
      echo "✓ Restart command successfully sent for task ${task}!"
    fi
  done
done

Automatic Retry Mechanisms #

To handle temporary (transient) network failures, we must not immediately move messages to the DLQ. We must give Connect workers the chance to automatically resend that data for a certain period.

Kafka Connect Core provides two main parameters for controlling retries:

  1. errors.retry.timeout: The maximum total time duration (in milliseconds) for a task to keep trying to resend a failed data batch before finally giving up and throwing an error. The default is 0 (no retry).
  2. errors.retry.delay.max.ms: The maximum pause time (in milliseconds) between retry calls. Connect uses an exponential backoff algorithm where the pause time keeps increasing on every failure until reaching the maximum limit set in this property (default: 60000 ms or 1 minute).

Below is a reliable retry tolerance setting example for facing short-lived target database downtime:

# Keep trying to resend data for a maximum of 5 minutes (300,000 ms) before the task is declared failed
errors.retry.timeout=300000

# Maximum pause between resends is 10 seconds
errors.retry.delay.max.ms=10000

Complete Sink Connector Configuration Example with Error Handling & DLQ #

Here’s a complete JSON configuration for deploying a production PostgreSQL Sink Connector equipped with transient error handling via retries, data format fault tolerance, and poison pill routing to a DLQ topic complete with metadata headers:

{
  "name": "postgresql-secure-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "2",
    "topics": "customer-transactions",
    "connection.url": "jdbc:postgresql://postgres-db:5432/finance",
    "connection.user": "sink_writer",
    "connection.password": "${file:/etc/connect/secrets:pg_writer_password}",
    
    "//": "--- Retry Configuration for Transient Network Errors ---",
    "//": "Keep trying to resend for 3 minutes",
    "errors.retry.timeout": "180000",
    "errors.retry.delay.max.ms": "5000",
    
    "//": "--- Data Format Fault Tolerance Configuration ---",
    "//": "all tolerance so the task doesn't crash when encountering corrupted JSON messages",
    "errors.tolerance": "all",
    
    "//": "--- Dead Letter Queue (DLQ) Configuration ---",
    "errors.deadletterqueue.topic.name": "dlq-customer-transactions",
    "//": "✓ CORRECT: Enabling error metadata header recording in the DLQ for debugging",
    "errors.deadletterqueue.context.headers.enable": "true",
    
    "//": "--- Database Write Settings ---",
    "insert.mode": "insert",
    "auto.create": "false",
    
    "//": "--- Converters ---",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

Summary #

  • Failure Classification — Errors in Kafka Connect are divided into transient network failures (need retries), data format errors / poison pills (need DLQ), and target logic errors.
  • Fail-Fast vs Tolerant — Use errors.tolerance=none for critical financial transaction systems for absolute consistency, and errors.tolerance=all for non-critical logs/metrics.
  • Dead Letter Queue — DLQ filters and isolates corrupted data into a separate Kafka topic, maintaining the operational continuity of distributed Connect worker tasks.
  • Metadata via Headers — Enable errors.deadletterqueue.context.headers.enable=true to embed the Exception class and original offset location in DLQ message headers to ease debugging processes.
  • Log4j & JMX Tuning — Monitor the total-records-skipped JMX metric and adjust the WorkerSinkTask logging package to DEBUG level for full visibility when errors occur.
  • DLQ Re-drive — Build a periodic DLQ reader application to fix data errors (poison pills) and send them back (re-ingest) to the main business topic.
  • Restart Automation — Deploy an automatic recovery shell script on the Connect REST API to restart tasks failed from transient network disruptions without manual intervention.

← Previous: Scaling & Resource Allocation Next: Connector Anti-Pattern →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact