Kafka Connect: JDBC Source, SMTs, Dead Letter Queues, and Connector Monitoring

Kafka Connect is Kafka's runtime for moving records between Kafka and external systems. The worker handles configuration, offsets, task assignment, and lifecycle; connector plugins implement a source or sink. Reliability depends on the worker mode, internal-topic durability, connector semantics, database cursor design, error policy, and the application's ability to reconcile duplicates or omissions.

Choose the worker mode deliberately

A standalone worker stores configuration and offsets locally. It can be appropriate for development or a controlled one-host process, but it has no automatic worker failover and local storage loss can remove its state. Distributed workers use Kafka topics for configuration, offsets, and task status, and rebalance work among available workers. That makes distributed mode the normal fault-tolerant choice.

For a Kafka cluster with at least three brokers, a starting worker configuration is:

bootstrap.servers=kafka-1:9093,kafka-2:9093,kafka-3:9093
group.id=connect-production

config.storage.topic=connect-production-configs
offset.storage.topic=connect-production-offsets
status.storage.topic=connect-production-status

config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider

The replication factor cannot exceed the number of brokers. Choose the highest value your failure model and broker count support; do not copy 3 into a smaller cluster. The configuration topic must have one partition and should be highly replicated. Protect internal topics with ACLs so only the Connect workers and administrators can modify them.

JDBC Source: cursor correctness before throughput

name=orders-jdbc-source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:postgresql://db.internal:5432/commerce
connection.user=connect_reader
connection.password=${file:/etc/kafka/secrets/database.properties:password}
table.include.list=public.orders
mode=timestamp+incrementing
timestamp.column.name=updated_at
incrementing.column.name=id
topic.prefix=db.orders.
poll.interval.ms=5000

The file provider must be enabled on every worker, and the referenced file must be readable only by the service account. A production secret manager provider can offer rotation and tighter controls.

The incrementing mode only finds larger cursor values, so it does not capture deletes and it misses updates to older rows. Timestamp mode depends on a reliably updated, comparable timestamp and must account for transactions that commit after their timestamp was assigned. Timestamp-plus-incrementing combines tie-breaking with time discovery, but it is still polling—not a database change log. Document the cursor invariant and test late commits, clock behavior, updates, deletes, and connector restarts.

The JDBC Source connector currently supports one task. If one poller cannot meet the workload, partition the source into intentionally disjoint connector queries or use a log-based CDC connector; do not assume tasks.max creates parallel table readers.

Debezium PostgreSQL for log-based CDC

name=orders-debezium
connector.class=io.debezium.connector.postgresql.PostgresConnector
topic.prefix=commerce
database.hostname=db.internal
database.port=5432
database.user=debezium
database.password=${file:/etc/kafka/secrets/database.properties:debezium_password}
database.dbname=commerce
plugin.name=pgoutput
slot.name=debezium_orders
publication.autocreate.mode=filtered
table.include.list=public.orders

Use topic.prefix; older examples that set database.server.name are not current. CDC latency and retention depend on database activity, connector health, Kafka availability, and consumer behavior, so do not promise a universal sub-second result. Monitor the replication slot and retained WAL as well as Connect task state. A stopped connector with a persistent slot can retain enough WAL to exhaust database storage.

Single Message Transforms

SMTs are appropriate for small, deterministic record-level changes such as routing a topic or extracting a field. They are not a replacement for joins, external lookups, or complex business logic.

transforms=route
transforms.route.type=org.apache.kafka.connect.transforms.RegexRouter
transforms.route.regex=db\.orders\.(.*)
transforms.route.replacement=orders-$1

Test transforms against tombstones, null fields, schema changes, and unexpected record types. A transform exception follows the connector's error policy and can stop or skip a task.

JDBC Sink: own the destination schema

name=orders-jdbc-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
topics=orders
connection.url=jdbc:postgresql://warehouse.internal:5432/reporting
connection.user=connect_writer
connection.password=${file:/etc/kafka/secrets/database.properties:warehouse_password}

auto.create=false
auto.evolve=false
insert.mode=upsert
pk.mode=record_key
pk.fields=id

Pre-create and migrate destination tables through a reviewed schema process. Automatic create/evolve is disabled by default and can produce unsuitable column types or allow a connector identity to run DDL. Upsert behavior depends on the destination dialect and a compatible primary key; validate duplicate delivery and replay before calling the sink idempotent.

Dead-letter queues are a controlled exception path

Kafka Connect's dead-letter queue support is for sink records that cannot be processed. Configure it only with an owner, retention policy, restricted ACLs, alerts, and a tested replay procedure:

errors.tolerance=all
errors.deadletterqueue.topic.name=orders-sink-dlq
errors.deadletterqueue.topic.replication.factor=3
errors.deadletterqueue.context.headers.enable=true
errors.log.enable=true
errors.log.include.messages=false

As with the worker's internal topics, the DLQ replication factor must fit the broker count. errors.tolerance=all means the connector can keep running while records are skipped, so connector health alone is insufficient. Alert on DLQ production and reconcile every record. Leave errors.log.include.messages=false unless a security review accepts writing record keys, values, or headers to logs.

Protect and use the REST API

The REST API can create, alter, restart, pause, and delete connectors. Expose it only on a management network with TLS and authentication, and avoid requesting expanded connector values when secrets could be returned.

curl --fail --silent --show-error --header "Authorization: Bearer $CONNECT_API_TOKEN" https://connect.example.internal:8443/connectors/orders-jdbc-source/status

A status response can be healthy while data freshness is not. Inspect connector state, every task state, the failure trace, source cursor progress, sink or consumer lag, DLQ ingress, database load, and end-to-end age of the newest usable record.

Metrics and recovery signals

• Alert on connector-failed-task-count and compare connector-running-task-count with connector-total-task-count.
• Track completed-rebalances-total; repeated rebalances can indicate unstable workers or connectivity.
• Use source-task and sink-task metrics for record rates, poll/batch behavior, and active counts.
• For downstream consumer groups, inspect lag with bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9093 --describe --group group-name.
• Test a worker loss, task exception, database outage, Kafka outage, schema change, credential rotation, replay, and DLQ reconciliation.

Official primary sources

• Apache Kafka 4.3 Connect user guide
• Apache Kafka 4.3 Connect worker configuration
• Apache Kafka 4.3 monitoring
• Confluent JDBC Source configuration
• Confluent JDBC Sink configuration
• Debezium PostgreSQL connector

Working with JusDB on Kafka database pipelines

JusDB reviews cursor correctness, CDC retention, destination constraints, connector security, failure handling, and end-to-end reconciliation for database-backed Kafka pipelines.

Explore JusDB Kafka services →  |  Talk to a database engineer