6.1 – Error Handling, Retries & Dead Letter Queues

Overview

This module is the authoritative guide to error handling, retries, and Dead Letter Queues (DLQs) in Kafka Connect. By the end, you will understand:

Retry behavior for transient vs. permanent errors.
Error tolerance configurations (none, all).
Dead Letter Queue (DLQ) topics for failed records.
Skip vs. fail strategies.
Poison record handling.

All error handling concepts live here. Other modules will reference this when discussing specific connectors or reliability patterns, but will not re-explain these mechanics.

Error Types in Kafka Connect

1. Transient Errors

Definition: Temporary failures that may succeed on retry.

Examples:

Network timeouts (e.g., database connection lost, Kafka broker unreachable).
Database deadlocks or temporary locking issues.
External API rate limiting (429 Too Many Requests).

Handling: Retry with exponential backoff.

2. Permanent Errors

Definition: Failures that will not succeed on retry.

Examples:

Invalid data format (e.g., malformed JSON, schema mismatch).
Constraint violations (e.g., database unique key violation).
Authorization failures (e.g., invalid credentials).

Handling: Skip or send to Dead Letter Queue (DLQ).

3. Poison Records

Definition: Records that consistently cause errors, preventing the task from progressing.

Examples:

A single malformed JSON record in a stream of valid records.
A record that violates database constraints.

Handling: Skip or send to DLQ to unblock the pipeline.

Kafka Connect Error Handling Architecture

Default Behavior (Strict)

By default, Kafka Connect uses a fail-fast strategy:

A task encounters an error.
The task stops and transitions to FAILED state.
The connector must be manually restarted to resume processing.

Use Case: Critical pipelines where data integrity is paramount.

Error Tolerance Mode

Kafka Connect supports an error tolerance mode that allows tasks to continue processing despite errors.

Configuration: errors.tolerance

Values:

none (default): Fail-fast (task stops on first error).
all: Continue processing, log errors, optionally send to DLQ.

Example:

{
  "errors.tolerance": "all"
}

Retry Configuration

1. Retry Timeout

Configuration: errors.retry.timeout

Purpose: Maximum time (in milliseconds) to retry a failed operation before giving up.

Default: 0 (no retries)

Example:

{
  "errors.retry.timeout": "300000"
}

Explanation: Retry for up to 5 minutes (300,000 ms) before marking the record as failed.

Best Practice: Set to 1-5 minutes for transient errors (e.g., network timeouts, temporary database unavailability).

2. Retry Delay

Configuration: errors.retry.delay.max.ms

Purpose: Maximum delay (in milliseconds) between retry attempts.

Default: 60000 (60 seconds)

Example:

{
  "errors.retry.delay.max.ms": "10000"
}

Explanation: Wait up to 10 seconds between retries (with exponential backoff).

Backoff Strategy: Connect uses exponential backoff starting from 1 second, doubling each retry until errors.retry.delay.max.ms is reached.

Example Backoff:

Retry 1: Wait 1 second
Retry 2: Wait 2 seconds
Retry 3: Wait 4 seconds
Retry 4: Wait 8 seconds
Retry 5+: Wait 10 seconds (capped at errors.retry.delay.max.ms)

Dead Letter Queue (DLQ)

What is a Dead Letter Queue?

A Dead Letter Queue (DLQ) is a Kafka topic where failed records are written for later inspection or reprocessing.

Purpose:

Unblock pipelines: Move poison records out of the main pipeline.
Audit failures: Inspect failed records to identify root causes.
Reprocess later: Manually or automatically retry failed records after fixing issues.

Enabling DLQ

Configuration: errors.deadletterqueue.topic.name

Example:

{
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "connect-dlq",
  "errors.deadletterqueue.topic.replication.factor": "3"
}

Explanation:

Failed records are written to the connect-dlq topic.
The DLQ topic is created with replication factor 3 for durability.

DLQ Record Format

DLQ records include metadata about the failure:

Key: Original record key (preserved).

Value: Original record value (preserved).

Headers (added by Connect):

__connect.errors.topic: Original Kafka topic.
__connect.errors.partition: Original Kafka partition.
__connect.errors.offset: Original Kafka offset.
__connect.errors.connector.name: Connector name.
__connect.errors.task.id: Task ID.
__connect.errors.stage: Stage where error occurred (TRANSFORMATION, VALUE_CONVERTER, HEADER_CONVERTER).
__connect.errors.exception.class.name: Exception class (e.g., org.apache.kafka.connect.errors.DataException).
__connect.errors.exception.message: Error message.
__connect.errors.exception.stacktrace: Full stack trace (if enabled).

Example (Consume DLQ):

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic connect-dlq \
  --from-beginning \
  --property print.key=true \
  --property print.headers=true

DLQ Context (Detailed Metadata)

Configuration: errors.deadletterqueue.context.headers.enable

Purpose: Include additional context in DLQ headers (e.g., stack traces, original topic metadata).

Default: false

Example:

{
  "errors.deadletterqueue.context.headers.enable": "true"
}

Impact: Larger DLQ messages (due to stack traces), but easier debugging.

Logging Errors

Log All Errors

Configuration: errors.log.enable

Purpose: Log all errors to the Connect worker logs (in addition to DLQ).

Default: false

Example:

{
  "errors.log.enable": "true",
  "errors.log.include.messages": "true"
}

Explanation:

All errors are logged to connect.log.
errors.log.include.messages=true: Include failed record content in logs (use with caution for sensitive data).

Best Practice: Enable for debugging, but disable in production (logs can grow large).

Error Handling Strategies

Strategy 1: Fail-Fast (Default)

Configuration:

{
  "errors.tolerance": "none"
}

Behavior:

Task stops on first error.
Operator must manually restart the connector.

Use Case: Critical pipelines where data integrity is paramount (e.g., financial transactions).

Pros: No data loss, no silent failures.
Cons: Requires manual intervention, pipeline stops on transient errors.

Strategy 2: Log and Continue

Configuration:

{
  "errors.tolerance": "all",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true"
}

Behavior:

Task continues processing despite errors.
Errors are logged, but records are not sent to DLQ (i.e., data is dropped).

Use Case: Non-critical pipelines where occasional data loss is acceptable (e.g., log ingestion).

Pros: Pipeline continues, no manual intervention.
Cons: Data loss, silent failures.

Strategy 3: Dead Letter Queue (Recommended)

Configuration:

{
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "connect-dlq",
  "errors.deadletterqueue.topic.replication.factor": "3",
  "errors.deadletterqueue.context.headers.enable": "true",
  "errors.log.enable": "true"
}

Behavior:

Task continues processing despite errors.
Failed records are written to DLQ.
Errors are logged.

Use Case: Production pipelines where data integrity is important, but downtime is unacceptable.

Pros: No pipeline downtime, failed records can be inspected and reprocessed.
Cons: Requires monitoring DLQ for failed records.

Strategy 4: Retry with DLQ Fallback

Configuration:

{
  "errors.tolerance": "all",
  "errors.retry.timeout": "300000",
  "errors.retry.delay.max.ms": "10000",
  "errors.deadletterqueue.topic.name": "connect-dlq",
  "errors.deadletterqueue.topic.replication.factor": "3"
}

Behavior:

Task retries transient errors for up to 5 minutes.
If retries fail, record is sent to DLQ.
Task continues processing.

Use Case: Pipelines with occasional transient errors (e.g., network timeouts, database connection issues).

Pros: Handles transient errors automatically, DLQ captures permanent failures.
Cons: Retries add latency for failed records.

Poison Record Handling

What is a Poison Record?

A poison record is a record that consistently causes errors, preventing the task from progressing (especially when errors.tolerance=none).

Example: A single malformed JSON record in a stream of valid records causes the connector to fail and stop processing all subsequent records.

Detection

Symptom: Connector repeatedly fails on the same offset.

Log Pattern:

Error processing record from topic 'orders' partition 2 offset 12345: 
org.apache.kafka.connect.errors.DataException: Failed to deserialize data

Mitigation Strategies

Option 1: Enable Error Tolerance + DLQ

Configuration:

{
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "connect-dlq"
}

Result: Poison record is sent to DLQ, connector continues processing.

Option 2: Skip the Record Manually

If DLQ is not enabled, you can manually advance the consumer offset:

Steps:

Pause the connector:

curl -X PUT http://localhost:8083/connectors/my-connector/pause

Manually commit offset past the poison record (requires custom script or Admin API).

Resume the connector:

curl -X PUT http://localhost:8083/connectors/my-connector/resume

Note: This is a last resort and should be automated via DLQ in production.

Option 3: Fix the Source Data

If the poison record originates from a source system, fix the data at the source and restart the connector.

Error Handling Best Practices

1. Always Use DLQ in Production

Rationale: Prevents pipeline downtime due to poison records.

Configuration:

{
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "connect-dlq"
}

2. Monitor DLQ for Failed Records

Tools:

Kafka Consumer: Consume DLQ and alert on new messages.
Prometheus + Grafana: Monitor DLQ topic lag.
Confluent Control Center: View DLQ messages in UI.

Alerting Example (Prometheus):

- alert: ConnectDLQRecords
  expr: kafka_topic_partition_current_offset{topic="connect-dlq"} > 0
  annotations:
    summary: "Kafka Connect DLQ has failed records"

3. Enable Logging for Debugging

Configuration:

{
  "errors.log.enable": "true",
  "errors.log.include.messages": "true"
}

Note: Disable errors.log.include.messages in production if records contain sensitive data.

4. Use Retries for Transient Errors

Configuration:

{
  "errors.retry.timeout": "300000",
  "errors.retry.delay.max.ms": "10000"
}

Rationale: Automatically recovers from temporary failures (e.g., network issues, database restarts).

5. Test Error Scenarios

Test Cases:

Malformed JSON input (deserialization failure).
Database constraint violation (duplicate key).
Network timeout (transient error).
External system unavailability (retry behavior).

Testing Tool: Chaos Engineering tools (e.g., Chaos Monkey) to simulate failures.

Example: Full Error Handling Configuration

Scenario: Robust Production Sink Connector

Requirements:

Retry transient errors for up to 5 minutes.
Send permanent failures to DLQ.
Log all errors for debugging.

Configuration:

{
  "name": "elasticsearch-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "3",
    "topics": "logs",
    "connection.url": "http://localhost:9200",
    
    "errors.tolerance": "all",
    "errors.retry.timeout": "300000",
    "errors.retry.delay.max.ms": "10000",
    "errors.deadletterqueue.topic.name": "connect-dlq",
    "errors.deadletterqueue.topic.replication.factor": "3",
    "errors.deadletterqueue.context.headers.enable": "true",
    "errors.log.enable": "true"
  }
}

Behavior:

Task encounters an error (e.g., Elasticsearch unavailable).
Task retries for up to 5 minutes with exponential backoff.
If error persists, record is sent to connect-dlq with full context.
Error is logged to connect.log.
Task continues processing subsequent records.

Common Error Scenarios

Scenario 1: Schema Mismatch

Error:

org.apache.kafka.connect.errors.DataException: Schema mismatch: expected field 'email', but was 'user_email'

Cause: Source data schema doesn't match target schema.

Solution:

Use SMTs to rename fields (ReplaceField).
Update target schema to match source.
Send to DLQ for manual inspection.

Scenario 2: Database Constraint Violation

Error:

org.postgresql.util.PSQLException: ERROR: duplicate key value violates unique constraint "users_pkey"

Cause: Attempting to insert a record with a duplicate primary key.

Solution:

Use upsert mode (insert.mode=upsert) in JDBC Sink.
Send to DLQ and manually resolve conflicts.

Scenario 3: Network Timeout

Error:

java.net.SocketTimeoutException: connect timed out

Cause: External system (database, API) is unreachable.

Solution:

Enable retries (errors.retry.timeout=300000).
If retries fail, send to DLQ.
Monitor external system health.

Key Takeaways

Error types: Transient (retry), permanent (skip/DLQ), poison records (unblock pipeline).
errors.tolerance: none (fail-fast) vs. all (continue processing).
Retries: Use errors.retry.timeout and errors.retry.delay.max.ms for transient errors.
DLQ: Send failed records to errors.deadletterqueue.topic.name for inspection.
Best practice: Enable DLQ + retries + logging in production.
Monitoring: Alert on DLQ topic lag to detect failures.

What's Next?

Module 7 covers custom connector development, including custom error handling logic.
Module 8.1 explains monitoring and alerting for connector failures.
Module 5.1 discusses performance tuning (retries add latency).

6.1 – Error Handling, Retries & Dead Letter Queues

Overview

Error Types in Kafka Connect

1. Transient Errors

2. Permanent Errors

3. Poison Records

Kafka Connect Error Handling Architecture

Default Behavior (Strict)

Error Tolerance Mode

Retry Configuration

1. Retry Timeout

2. Retry Delay

Dead Letter Queue (DLQ)

What is a Dead Letter Queue?

Enabling DLQ

DLQ Record Format

DLQ Context (Detailed Metadata)

Logging Errors

Log All Errors

Error Handling Strategies

Strategy 1: Fail-Fast (Default)

Strategy 2: Log and Continue

Strategy 3: Dead Letter Queue (Recommended)

Strategy 4: Retry with DLQ Fallback

Poison Record Handling

What is a Poison Record?

Detection

Mitigation Strategies

Option 1: Enable Error Tolerance + DLQ

Option 2: Skip the Record Manually

Option 3: Fix the Source Data

Error Handling Best Practices

1. Always Use DLQ in Production

2. Monitor DLQ for Failed Records

3. Enable Logging for Debugging

4. Use Retries for Transient Errors

5. Test Error Scenarios

Example: Full Error Handling Configuration

Scenario: Robust Production Sink Connector

Common Error Scenarios

Scenario 1: Schema Mismatch

Scenario 2: Database Constraint Violation

Scenario 3: Network Timeout

Key Takeaways

What's Next?

Further Reading