> ## Content Index > Fetch the complete content index at: https://www.ggorantala.dev/llms.txt > Use this file to discover other available public pages before exploring further. # Kafka Connect Task Threw an Uncaught and Unrecoverable Exception - URL: https://www.ggorantala.dev/kafka-connect-task-threw-uncaught-unrecoverable-exception/ - Published: 2026-03-01T14:25:00.000Z - Updated: 2026-08-29T08:20:20.000Z - Description: Kafka Connect task killed with 'uncaught and unrecoverable exception'? It's usually a converter poison pill. Here's the DataException root cause, DLQ fix, and design. - Author: Gopi Gorantala - Tags: kafka-connect, kafka-errors, ConnectException, DataException, dead-letter-queue, kafka-docker, event-streaming Your connector shows `RUNNING`, then a task flips to `FAILED` and stays there. Nothing recovers on its own, throughput on that partition set drops to zero, and consumer lag on the source topic climbs. This is the single most-pasted Kafka Connect failure, and 90% of the time it is a converter poison pill — one bad record that the deserializer can't parse, killing the whole task. ## 1\. The Error The line everyone searches for, from the worker log: ```text ERROR WorkerSinkTask{id=jdbc-sink-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. Error: Converting byte[] to Kafka Connect data failed due to serialization error: (org.apache.kafka.connect.runtime.WorkerSinkTask) org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error: at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:333) at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87) at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$4(WorkerSinkTask.java:539) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:180) at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:214) ... Caused by: org.apache.kafka.common.errors.SerializationException: java.io.CharConversionException: Invalid UTF-32 character 0x1876f6d(above 0x0010ffff) at char #1, byte #7) ``` Or, the Avro/Schema-Registry twin that shows up when a `JsonConverter` is pointed at a topic written by an Avro serializer: ```text org.apache.kafka.connect.errors.DataException: Converting byte[] to Kafka Connect data failed due to serialization error: Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte! ``` The status API confirms the task is dead, not just paused: ```bash curl -s localhost:8083/connectors/jdbc-sink/status | jq ``` ```json { "name": "jdbc-sink", "connector": { "state": "RUNNING", "worker_id": "connect:8083" }, "tasks": [ { "id": 0, "state": "FAILED", "worker_id": "connect:8083", "trace": "org.apache.kafka.connect.errors.ConnectException: Exit the WorkerSinkTask due to unrecoverable exception..." } ] } ``` Note the split-brain: **connector `RUNNING`, task `FAILED`**. The connector object is just config; the task is what actually moves data. A `RUNNING` connector with `FAILED` tasks moves nothing. **Environment:** Apache Kafka Connect 2.0+ (the error-handling framework is KIP-298, shipped in 2.0). Reproductions below use `apache/kafka:3.9.0` and Confluent's `cp-kafka-connect:7.9.x` / `kafka-connect-jdbc`. Applies identically to KRaft-mode clusters — Connect is a client, it doesn't care about ZooKeeper vs KRaft. ## 2\. How to Reproduce It Stand up a broker plus a Connect worker with the JSON converters wired as defaults: ```yaml # docker-compose.yml services: kafka: image: apache/kafka:3.9.0 ports: ["9092:9092"] environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092 KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 connect: image: confluentinc/cp-kafka-connect:7.9.0 depends_on: [kafka] ports: ["8083:8083"] environment: CONNECT_BOOTSTRAP_SERVERS: kafka:9092 CONNECT_GROUP_ID: connect-cluster CONNECT_CONFIG_STORAGE_TOPIC: _connect-configs CONNECT_OFFSET_STORAGE_TOPIC: _connect-offsets CONNECT_STATUS_STORAGE_TOPIC: _connect-status CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1 CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1 CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1 CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false" CONNECT_PLUGIN_PATH: /usr/share/java,/usr/share/confluent-hub-components ``` Create a topic and a file-sink connector (no external DB needed to trigger the converter failure): ```bash docker compose up -d sleep 20 docker compose exec kafka /opt/kafka/bin/kafka-topics.sh \ --create --topic events --bootstrap-server kafka:9092 \ --partitions 1 --replication-factor 1 curl -s -X PUT localhost:8083/connectors/file-sink/config \ -H 'Content-Type: application/json' -d '{ "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector", "tasks.max": "1", "topics": "events", "file": "/tmp/out.txt" }' ``` Now write one good record, then one poison pill straight from the console producer (raw bytes, no schema): ```bash # valid JSON — processed fine echo '{"id":1,"name":"ok"}' | docker compose exec -T kafka \ /opt/kafka/bin/kafka-console-producer.sh --topic events --bootstrap-server kafka:9092 # poison pill — not JSON at all echo 'this-is-not-json' | docker compose exec -T kafka \ /opt/kafka/bin/kafka-console-producer.sh --topic events --bootstrap-server kafka:9092 ``` Within a second, the task dies with `DataException: Converting byte[] to Kafka Connect data failed`. Check it: ```bash curl -s localhost:8083/connectors/file-sink/status | jq '.tasks[0].state' # "FAILED" ``` The `**Unknown magic byte!**` variant reproduces just as easily: produce with `kafka-avro-console-producer` (which prepends the 5-byte magic-byte + schema-ID header) while the connector uses `JsonConverter`. The JSON parser hits byte `0x00` and blows up. ## 3\. Why It Happens — Surface Level A sink task's inner loop is `poll() → convert → transform → put()`. The **converter** runs first, turning the raw `byte[]` value from the topic into a Connect `SchemaAndValue`. If the bytes aren't the shape the converter expects — non-JSON bytes under `JsonConverter`, or Avro bytes under `JsonConverter`, or JSON-with-schema when `schemas.enable=false` — `toConnectData()` throws a `DataException`. By default `errors.tolerance=none`. Any exception in the convert/transform stage is treated as fatal: the `RetryWithToleranceOperator` gives up, the `WorkerSinkTask` catches it, logs "uncaught and unrecoverable," wraps it in a `ConnectException`, and terminates. One malformed record on one partition kills the entire task and every partition it owned. ## 4\. Why It Happens — Under the Hood KIP-298 ("Error Handling in Connect," Kafka 2.0) introduced the `RetryWithToleranceOperator`, which wraps each stage that can fail on a per-record basis: the key converter, the value converter, the header converter, and each Single Message Transform (SMT). Two knobs govern it: - `errors.retry.timeout` (default `0`) — how long to keep retrying a **retriable** failure before giving up. Converter deserialization errors are *not* retriable, so retries never apply to a poison pill; the same bytes fail identically every time. - `errors.tolerance` (default `none`) — what to do once retries are exhausted. `none` \= rethrow and kill the task. `all` \= increment a metric, optionally route to the DLQ, and move on. Crucially, **the DLQ only captures failures in the converter and transform stages** — because those failures happen inside the framework, which still holds the original `SinkRecord` bytes and can re-produce them to another topic. Failures inside the connector's own `put()` (a JDBC unique-constraint violation, an Elasticsearch mapping conflict) are also governed by `errors.tolerance`, but there is **no DLQ for them** — the framework no longer has a routable original record, so a tolerated `put()` failure is logged/skipped, not dead-lettered. This asymmetry trips up a lot of teams: they set a DLQ, watch it stay empty while records vanish, and assume it's broken. Also understand the blast radius. `tasks.max` splits a topic's partitions across tasks. A single poison record kills *that task*, which owns a slice of partitions — so a 12-partition topic with `tasks.max=3` loses a third of its throughput per dead task, not one partition. And because the sink task commits offsets only after a successful `put()` batch, restarting a task with the poison pill still in place just re-reads and re-dies. Manual restart without a tolerance change is a loop. ## 5\. The Fix ### Immediate: tolerate errors and route to a DLQ (sink connectors) ```diff { "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": "3", "topics": "events", + "errors.tolerance": "all", + "errors.deadletterqueue.topic.name": "events.dlq", + "errors.deadletterqueue.topic.replication.factor": "1", + "errors.deadletterqueue.context.headers.enable": "true", + "errors.log.enable": "true", + "errors.log.include.messages": "true" } ``` `errors.tolerance=all` stops the task from dying. `errors.deadletterqueue.topic.name` sends the offending raw record to a side topic instead of dropping it. **Always** set `errors.deadletterqueue.context.headers.enable=true` — without it you get the bad bytes in the DLQ but no reason *why* they failed. With it, each DLQ record carries headers like: ```text __connect.errors.topic __connect.errors.partition __connect.errors.offset __connect.errors.exception.class.name __connect.errors.exception.message __connect.errors.exception.stacktrace ``` Inspect them: ```bash docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server kafka:9092 --topic events.dlq \ --from-beginning --property print.headers=true --max-messages 5 ``` Set `errors.deadletterqueue.topic.replication.factor=3` in any real cluster (the `1` above is dev-only). ### Root cause: use the converter that matches the data If the topic is Avro, use the Avro converter — don't tolerate a mismatch you can fix: ```diff - "value.converter": "org.apache.kafka.connect.json.JsonConverter", - "value.converter.schemas.enable": "false" + "value.converter": "io.confluent.connect.avro.AvroConverter", + "value.converter.schema.registry.url": "http://schema-registry:8081" ``` `Unknown magic byte!` almost always means "you're reading Avro/Protobuf bytes with a JSON or String converter." Fix the converter, not the tolerance. ### Which fix when - **Schema/format mismatch across the whole topic** (every record fails): wrong converter. DLQ would just mirror the entire topic — fix the converter. - **Occasional bad records in an otherwise-good stream**: `errors.tolerance=all` \+ DLQ. This is the poison-pill case the DLQ was built for. - **`put()`\-side failures** (constraint violations, mapping errors): `errors.tolerance=all` skips them (logged, not dead-lettered); prefer fixing the upstream data or connector config so you're not silently dropping rows. ## 6\. Best Practices & The Better Design The fundamentally better setup treats bad data as expected, not exceptional. Every sink connector in production ships with this block from day one: ```json { "errors.tolerance": "all", "errors.deadletterqueue.topic.name": "${connector}.dlq", "errors.deadletterqueue.topic.replication.factor": "3", "errors.deadletterqueue.context.headers.enable": "true", "errors.log.enable": "true", "errors.log.include.messages": "true", "errors.retry.timeout": "300000", "errors.retry.delay.max.ms": "60000" } ``` `errors.retry.timeout=300000` (5 min) with backoff means *transient* failures — a momentarily unavailable DB, a Schema Registry blip — retry instead of dead-lettering good records. Poison pills, being non-retriable, still go straight to the DLQ. You get retries where they help and tolerance where they don't. Beyond config, enforce format at the boundary. The reason a poison pill exists at all is that *some producer wrote bytes that don't match the topic's contract*. Register schemas in a Schema Registry and set producers to `auto.register.schemas=false` with compatibility checks, so malformed or incompatible records are rejected at write time — the sink never sees them. A DLQ is a safety net, not a substitute for a schema contract on the topic. Give each connector its **own** DLQ topic (`${connector}.dlq`), never a shared one — mixing failures from ten connectors into one topic makes triage miserable and couples their retention. And wire a small consumer (or a second Connect job) on the DLQ that alerts and, once the bug is fixed, replays corrected records back to the source topic. ## 7\. How to Prevent It Long-Term Monitoring is non-negotiable because a `FAILED` task is silent — nothing throws in *your* app, lag just grows. Track: - **Connector/task state** — poll `GET /connectors/{name}/status` (or the `kafka.connect:type=connector-task-metrics` `status` metric) and alert on any task not `RUNNING`. This is the single most important Connect alert. - **DLQ record rate** — a nonzero, rising DLQ is your early warning that a producer changed format. Alert on DLQ produce rate > 0. - **`RetryWithToleranceOperator` metrics** — `deadletterqueue-produce-requests`, `total-record-errors`, `total-records-skipped`, `total-record-failures` are exposed under `kafka.connect:type=task-error-metrics`. Skipped records climbing means you're silently dropping data. Team conventions that catch this before prod: standardize the error-handling block above in your connector config templates (Terraform/GitOps), so no connector ships with `errors.tolerance=none` by accident. Add a CI smoke test with Testcontainers that stands up Connect, produces one valid and one poison record, and asserts the task stays `RUNNING` and the DLQ received exactly one message. And pin converter choice to the topic's registered format via review — a `JsonConverter` on an Avro topic should never pass code review. ## 8\. Key Takeaways - **`Task threw an uncaught and unrecoverable exception` \= a poison pill killed the task** — usually a converter `DataException` (`Unknown magic byte!` means Avro/Protobuf bytes read by a JSON/String converter). - **Connector `RUNNING` \+ task `FAILED` moves zero data.** Always check `GET /connectors/{name}/status`, not just the connector state, and alert on any non-`RUNNING` task. - **`errors.tolerance=all` \+ a DLQ** stops one bad record from killing the task — but the DLQ only captures *converter/transform* failures, not `put()`\-side failures. - **Always enable `errors.deadletterqueue.context.headers.enable=true`**, or your DLQ tells you *what* failed but never *why*. - **A DLQ is a net, not a fix.** Enforce the format at the producer with a Schema Registry contract so poison pills never reach the topic.