Kafka Connect in Production: Converters, Dead Letter Queues, and Worker Sizing

Kafka Connect has a reputation for being the easy part of Kafka. You post some JSON at a REST endpoint, a connector appears, and data moves. That reputation holds for exactly as long as your data is clean, your schemas are stable, and nobody restarts a worker.

The first Connect cluster I inherited had eleven connectors, four of which were in FAILED state and had been for weeks. Nobody noticed, because the health check was on the worker process and the worker was fine. The tasks were dead. One had hit a single record with an unexpected null in a non-optional field; the others had all failed for reasons the team never diagnosed because the stack trace mentioned DataException and stopped being read there.

This post is the configuration and operational discipline I now apply to every Connect deployment: how to pick converters so serialization stops being a mystery, how to make errors visible and survivable with a dead letter queue, where single message transforms stop being appropriate, and how to size workers and tasks so a scale-up actually scales. If Kafka’s basics are still settling, the fundamentals post is the prerequisite.

The architecture in four nouns

A worker is a JVM process. In distributed mode, workers with the same group.id form a cluster, coordinate through Kafka’s group protocol, and store their state in three internal topics: config, offsets, and status.

A connector is a configuration object. It does no data movement itself; it decides how to split the work.

A task is the unit that actually copies data. Connect distributes tasks across workers. A sink connector’s tasks each own a subset of topic partitions, exactly like a consumer group. A source connector’s task count is capped by whatever the connector can parallelize — a JDBC source splits by table, a Debezium Postgres source is capped at one because there is one replication slot.

A converter turns Connect’s internal record representation into bytes and back. This is the piece that causes the most production pain, and it is the piece with the least intuitive error messages.

flowchart TD
  src["Source system"] --> stask["Source task"]
  stask --> smt1["SMTs"]
  smt1 --> conv1["Converter (serialize)"]
  conv1 --> topic["Kafka topic"]
  topic --> conv2["Converter (deserialize)"]
  conv2 --> smt2["SMTs"]
  smt2 --> ktask["Sink task"]
  ktask --> dest["Warehouse or index"]
  conv2 -->|failure| dlq["Dead letter queue topic"]
  smt2 -->|failure| dlq

Converters sit at both edges. The DLQ only catches sink-side failures, which is a limitation worth remembering.

Converters: pick one and be explicit

The three converters that matter in practice:

AvroConverter (or ProtobufConverter / JsonSchemaConverter) with Schema Registry. Compact, schema-enforced, and the right default for anything internal. Pairs with the compatibility discipline in the schema registry post.

JsonConverter with schemas.enable=true. Every message carries its own schema inline, which roughly triples the payload and makes the topic unreadable to humans anyway. Avoid.

JsonConverter with schemas.enable=false, or StringConverter, or ByteArrayConverter. Schemaless. Fine for a sink that does its own parsing, and required when reading a topic written by something that never heard of Connect.

The failure that eats afternoons: a sink connector configured with AvroConverter pointed at a topic written by a plain JSON producer. The error is Unknown magic byte! or an Invalid UTF-8 complaint deep in a deserializer, and neither says “your converter does not match the data on this topic.”

Set converters explicitly per connector rather than relying on the worker default, because the worker default will be right for nine connectors and wrong for the tenth:

{
  "name": "orders-to-snowflake",
  "config": {
    "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
    "topics": "orders.placed,orders.updated",
    "tasks.max": "4",

    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.auto.register.schemas": "false",

    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq.orders-to-snowflake",
    "errors.deadletterqueue.topic.replication.factor": "3",
    "errors.deadletterqueue.context.headers.enable": "true",
    "errors.log.enable": "true",
    "errors.log.include.messages": "false",
    "errors.retry.timeout": "300000",
    "errors.retry.delay.max.ms": "30000",

    "consumer.override.max.poll.records": "500",
    "consumer.override.isolation.level": "read_committed"
  }
}

Note consumer.override.*. Sink connectors run consumers, and every consumer tuning concern from the rebalancing post applies — including max.poll.records sizing against the sink’s write latency. Connect exposes those through the override prefix, and the worker must permit it with connector.client.config.override.policy=All.

Dead letter queues, and their sharp edge

errors.tolerance defaults to none, which means the first record that cannot be converted or transformed kills the task. The worker stays healthy. Your dashboard stays green. Data stops moving.

Setting errors.tolerance=all with a DLQ topic changes that: bad records go to the DLQ with headers describing what went wrong, and the task keeps running.

The sharp edge is scope. The DLQ catches failures in deserialization, conversion, and SMTs. It does not catch failures inside the connector’s own write to the destination. A Snowflake sink that fails because the target table is missing a column does not produce DLQ records; it retries and eventually fails the task. So the DLQ is necessary and not sufficient — you still need task-state monitoring.

errors.deadletterqueue.context.headers.enable=true is the flag that makes the DLQ usable. Without it you get the raw bad bytes and no explanation. With it, each DLQ record carries headers naming the original topic, partition, offset, the exception class, and the stage that failed:

"""Triage DLQ records: group failures by cause before fixing anything."""
from collections import Counter

from confluent_kafka import Consumer

consumer = Consumer({
    "bootstrap.servers": "b-1.msk.internal:9092",
    "group.id": "dlq-triage",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["dlq.orders-to-snowflake"])

causes: Counter[str] = Counter()
samples: dict[str, tuple[str, int, int]] = {}

while (msg := consumer.poll(5.0)) is not None:
    headers = {k: (v or b"").decode("utf-8", "replace") for k, v in (msg.headers() or [])}
    cause = "{}: {}".format(
        headers.get("__connect.errors.exception.class.name", "unknown"),
        headers.get("__connect.errors.exception.message", "")[:120],
    )
    causes[cause] += 1
    samples.setdefault(cause, (
        headers.get("__connect.errors.topic", "?"),
        int(headers.get("__connect.errors.partition", -1)),
        int(headers.get("__connect.errors.offset", -1)),
    ))

for cause, count in causes.most_common(10):
    topic, partition, offset = samples[cause]
    print(f"{count:>7}  {cause}\n         first at {topic}[{partition}]@{offset}")

Run that before you write a single fix. In my experience a DLQ with 400,000 records has two distinct causes, and one of them is usually a single upstream field that started arriving null.

A DLQ that nobody reads is a slower version of dropping the data. I alert on DLQ topic message rate above zero for more than five minutes, and I set retention on DLQ topics to at least a week so replay is possible after the fix.

SMTs: useful, and smaller than you think

Single message transforms run in-process, per record, with no state and no external calls. They are excellent for the reshaping that would otherwise require a whole stream processing job:

{
  "transforms": "unwrap,route,ts,mask",

  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.drop.tombstones": "false",
  "transforms.unwrap.add.fields": "op,source.ts_ms:source_ts_ms",

  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "cdc\\.public\\.(.*)",
  "transforms.route.replacement": "raw_$1",

  "transforms.ts.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
  "transforms.ts.field": "created_at",
  "transforms.ts.target.type": "Timestamp",
  "transforms.ts.format": "yyyy-MM-dd HH:mm:ss",

  "transforms.mask.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.mask.fields": "ssn,card_number",
  "transforms.mask.replacement": "****"
}

Transforms apply in the order named in transforms, and that order matters: unwrapping the Debezium envelope has to happen before any transform that references a business field, because before the unwrap those fields live under after.

Where SMTs stop being the right tool:

• Anything needing a lookup against another topic or a database. That is a join, it needs state, and it belongs in Kafka Streams or Flink.
• Anything needing to aggregate or window. SMTs see one record.
• Long chains. Six or seven transforms per record is a real CPU cost at high throughput and an untestable pile of JSON. Past four, I write a stream processing job.
• Conditional logic beyond what the Filter and Predicate machinery expresses. Nested predicates in JSON are worse than ten lines of code.

Sizing workers and tasks

tasks.max is a request, not a promise. Connect grants the minimum of what you asked for and what the connector can parallelize:

• Sink connectors: capped by the number of partitions across the subscribed topics. tasks.max=8 on a four-partition topic gives you four working tasks and four idle ones.
• JDBC source: capped by the number of tables in the query set.
• Debezium Postgres source: capped at 1, because one replication slot has one ordered stream. Scaling means more connectors on different tables, not more tasks. Details in the Debezium guide.
• S3 / file source: capped by object or partition enumeration.

Workers are where memory and CPU live. My starting point for a production cluster: three workers, 4 GB heap each, rack.aware.assignment where the cloud supports it, and a target of no more than roughly 15 tasks per worker for sinks that buffer in memory. Connect’s rebalance protocol is the incremental cooperative one by default in recent versions (connect.protocol=sessioned), so adding a worker does not stop the world — but a connector reconfiguration still restarts that connector’s tasks.

Heap sizing is driven by the sinks that batch. A Snowflake or S3 sink holding a 128 MB buffer per task across 12 tasks needs real memory, and the failure mode is a GC pause long enough to trip the consumer’s poll interval, which looks like a rebalance problem and is actually a sizing problem.

flowchart LR
  w1["Worker 1"] --> t1["Task 0-3"]
  w2["Worker 2"] --> t2["Task 4-7"]
  w3["Worker 3"] --> t3["Task 8-11"]
  cfg["Config topic"] --> w1
  cfg --> w2
  cfg --> w3
  off["Offsets topic"] --> w1
  st["Status topic"] --> mon["Monitoring"]

Three internal topics hold all cluster state. Back them up; a lost config topic is a lost cluster.

Configuration as code, not as REST calls

The Connect REST API makes it trivially easy to change production by hand, and that is its worst property. A connector config edited through curl at 4 p.m. exists in exactly one place: the internal config topic. There is no diff, no review, and no record of who changed tasks.max from 4 to 12.

I keep every connector config as a JSON file in git and apply it from CI with PUT /connectors/{name}/config, which is idempotent and creates-or-updates. Secrets come from the FileConfigProvider or a vault provider so they never appear in the file or in a GET response:

{
  "connection.password": "${file:/opt/connect/secrets.properties:snowflake_pk_passphrase}"
}

Two operational habits that go with this. First, the three internal topics (connect-configs, connect-offsets, connect-status) must be compacted with replication factor 3; connect-configs in particular must be single-partition, and losing it means rebuilding every connector from scratch. Second, monitor task state, not worker state:

# Anything not RUNNING is an incident. Wire this into your alerting.
curl -s http://connect:8083/connectors?expand=status \
  | jq -r '.[] | .status | .name as $n
           | .tasks[] | select(.state != "RUNNING")
           | "\($n) task \(.id) \(.state): \(.trace | split("\n")[0])"'

A failed task does not restart itself. POST /connectors/{name}/tasks/{id}/restart brings it back, and a wrapper that does this automatically for transient failures is worth writing — but only after you have the DLQ in place, otherwise you are restarting into the same poison record forever.

Pitfalls

Converter mismatch. The single most common Connect failure and the one with the least helpful error text. Set converters per connector and verify against what is actually on the topic before deploying.

errors.tolerance=none in production. One bad record stops a task permanently. The worker health check stays green while data stops flowing.

A DLQ nobody monitors. Records land, the task keeps running, and six weeks later someone asks why the warehouse is missing 3% of orders.

tasks.max above the partition count. Idle tasks consume worker slots and give the illusion of parallelism. Match tasks to the actual parallelism ceiling of the connector.

Editing config through the REST API by hand. No history, no review, and the change survives only in the config topic until someone overwrites it from a stale file.

Ignoring the internal topics. Under-replicated or multi-partition connect-configs is a cluster you cannot recover. Set replication factor 3, compaction, and one partition.

Heavy SMT chains at high throughput. Every transform is per-record CPU in the worker JVM. Six transforms at 50,000 records per second is a real bill, and it is invisible until the worker starts pausing.

FAQ

Should I run Connect on Kubernetes or as a managed service?

Managed Connect removes the worker operations and constrains connector choice and plugin versions. I run self-managed when I need a specific connector version or custom SMTs, and managed otherwise. Either way the connector config discipline in this post is unchanged.

How do I handle schema changes in a sink connector?

Depends on the sink. JDBC and Snowflake sinks can auto-evolve the target table when auto.evolve is on, which handles additive columns and nothing else. Breaking changes still need a new topic and a new connector, which is why the compatibility mode on the subject matters as much as the connector config.

Can Connect do exactly-once?

Source connectors support exactly-once with exactly.once.source.support=enabled on the worker, which uses Kafka transactions. Sink connectors are at-least-once unless the specific connector implements upserts on a key. Assume duplicates and make the sink idempotent, as argued in the exactly-once post.

What is the right retention for a DLQ topic?

At least a week, ideally matching the source topic. The DLQ is your replay source after a fix, and a three-day retention means a bug found on Friday is unfixable on Monday.

Why did adding a worker not increase throughput?

Because tasks are the unit of parallelism and you probably did not add tasks. Check the actual running task count in the status endpoint against tasks.max and against the connector’s ceiling.

Should sink connectors and source connectors share a cluster?

I separate them. Source connectors are usually latency-sensitive and low-volume; sinks are throughput-heavy and memory-hungry. A Debezium connector sharing heap with an S3 sink that buffers 128 MB per task is a bad afternoon waiting for a traffic spike.

What this means for your pipelines

Connect earns its place when the alternative is writing and operating a bespoke consumer for every destination. The catch is that its defaults are tuned for a demo, not for a pipeline with an SLA. Three configuration changes do most of the work: explicit converters per connector, errors.tolerance=all with a monitored DLQ, and consumer.override.max.poll.records sized against your sink’s write latency.

After that, the discipline is operational rather than technical. Connector configs belong in git and get applied by CI, so a tasks.max change is a reviewable diff instead of a curl somebody half-remembers running. Monitoring watches task state, not worker state, because the worker will happily report itself healthy while every task under it is dead. And the DLQ gets a real alert with a real owner, because an error channel nobody reads is just a slower way to lose records.

Get those in place and Connect becomes genuinely boring infrastructure, which is the highest praise I have for a data movement tool. The connectors that remain hard — CDC in particular — are hard for reasons specific to the source system, and that is where your attention should go instead.