Monitoring and Operating Kafka Connect Clusters

Running Apache Kafka Connect in production demands disciplined observability, lifecycle management, and failure remediation. This guide targets platform and SRE engineers responsible for distributed Kafka Connect clusters. It covers the metrics, logs, REST operations, and automation patterns that keep data flowing reliably, with every MBean name, config key, and REST endpoint verified against the Apache Kafka and Confluent documentation. We assume familiarity with Kafka Connect Architecture and Deployment Modes, including distributed vs. standalone modes, the role of workers, and internal topic-based configuration storage.

Instrumenting the Connect Cluster

Kafka Connect exposes metrics via Java Management Extensions (JMX). The metric groups that matter operationally are worker health, rebalance behavior, and per-task throughput. The exact MBean names below come from the Connect metrics reference (KIP-196); using the wrong domain or attribute name is a common cause of silently broken dashboards.

Worker and Rebalance Metrics

Worker-level health and group-coordination behavior live in two separate MBeans. Note that rebalance timing is under connect-worker-rebalance-metrics, not a connect-coordinator-metrics bean — there is no such bean in Connect.

A drop in task-count without a matching rise in completed-rebalances-total indicates tasks moving to FAILED rather than being reassigned. Frequent full GCs cause worker unresponsiveness, missed group heartbeats, and avoidable rebalances.

Connector and Task Metrics

The status attribute on connector-task-metrics is a string (running, paused, failed, unassigned, destroyed); alert when it is not running. For sink connectors, a persistent gap between sink-record-read-rate (records read from Kafka) and sink-record-send-rate (records sent downstream) signals an under-provisioned or stalled sink.

For end-to-end sink lag, sink-task-metrics exposes sink-record-lag-max (the maximum lag in records for any topic partition the task consumes). Cluster-wide consumer lag is still best confirmed out-of-band with kafka-consumer-groups, because the per-task gauge disappears when a task is FAILED:

kafka-consumer-groups --bootstrap-server kafka1:9092 \
  --describe --group connect-pg-source-inventory

Connect names each sink connector’s consumer group connect-<connector-name>.

Exposing Metrics to Prometheus

There is no built-in Prometheus reporter in Apache Kafka Connect. The standard approach is the Prometheus JMX Exporter Java agent, attached via KAFKA_OPTS so it loads in the same JVM:

export KAFKA_OPTS="-javaagent:/opt/jmx_prometheus_javaagent.jar=8084:/opt/connect-jmx-exporter.yml"

The agent serves Prometheus metrics on port 8084; no remote JMX port or com.sun.management.jmxremote flags are required, since the agent reads MBeans in-process. A minimal exporter config:

rules:
  - pattern: "kafka.connect<type=connect-worker-metrics><>(connector-count|task-count)"
    name: kafka_connect_worker_$1
  - pattern: "kafka.connect<type=connect-worker-rebalance-metrics><>(rebalance-avg-time-ms|completed-rebalances-total)"
    name: kafka_connect_rebalance_$1
  - pattern: "kafka.connect<type=connector-task-metrics, connector=(.+), task=(.+)><>(status|batch-size-avg|offset-commit-avg-time-ms)"
    name: kafka_connect_task_$3
    labels:
      connector: "$1"
      task: "$2"
  - pattern: "kafka.connect<type=sink-task-metrics, connector=(.+), task=(.+)><>(sink-record-read-rate|sink-record-send-rate|sink-record-lag-max)"
    name: kafka_connect_sink_$3
    labels:
      connector: "$1"
      task: "$2"

The status attribute is a string, so JMX Exporter emits it as a labeled metric value of 1; alert on its presence/absence per state rather than a numeric threshold.

Alerting Rules

groups:
- name: kafka-connect
  rules:
  - alert: ConnectTaskFailed
    expr: kafka_connect_task_status{status="failed"} > 0
    for: 2m
    labels:
      severity: critical
    annotations:
      summary: "Connect task {{ $labels.connector }}/{{ $labels.task }} is FAILED"
  - alert: ConnectSinkLagHigh
    expr: kafka_connect_sink_sink_record_lag_max > 10000
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Sink connector {{ $labels.connector }} lag exceeds 10k records"

Pick the lag threshold from the connector’s own throughput, not a round number: if a sink sustains 2,000 records/s, a 10k backlog is ~5 s of catch-up and rarely worth a page, whereas the same 10k on a 50 records/s connector is over three minutes behind. Express the threshold as sink-record-lag-max > N * sink-record-send-rate where N is your tolerated seconds-behind.

Logging Configuration and Diagnostics

Apache Kafka migrated its logging backend from Log4j 1.x to Log4j2 in Kafka 4.0 (KIP-653); the default Connect config file became connect-log4j2.properties. Releases before 4.0 use Log4j 1.x with connect-log4j.properties. Match the syntax to your running version. (Confluent Platform ships a YAML variant, connect-log4j2.yaml.)

Routing Connector Logs (Log4j 1.x, Kafka < 4.0)

# connect-log4j.properties
log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ISO8601} [%t] %-5p %c{2} - %m%n

log4j.logger.org.apache.kafka.connect.runtime.WorkerSinkTask=DEBUG
log4j.logger.org.apache.kafka.connect.runtime.WorkerSourceTask=DEBUG

For Kafka >= 4.0, the equivalent in connect-log4j2.properties:

# connect-log4j2.properties
appender.console.type = Console
appender.console.name = STDOUT
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d{ISO8601} [%t] %-5p %c{2} - %m%n

rootLogger.level = INFO
rootLogger.appenderRef.console.ref = STDOUT

logger.sinktask.name = org.apache.kafka.connect.runtime.WorkerSinkTask
logger.sinktask.level = DEBUG
logger.sourcetask.name = org.apache.kafka.connect.runtime.WorkerSourceTask
logger.sourcetask.level = DEBUG

Dynamic Log Levels via REST

Change levels at runtime without restarting the worker (the admin/loggers endpoint exists since Kafka 2.4, KIP-495):

curl -s -X PUT -H "Content-Type: application/json" \
  -d '{"level": "DEBUG"}' \
  "http://connect-worker:8083/admin/loggers/org.apache.kafka.connect.runtime.WorkerSinkTask"

The response lists the loggers whose level changed. By default the change applies to the local worker only (scope Worker); pass ?scope=cluster (Kafka 3.7+, KIP-976) to propagate it to every worker via the config topic — a cluster-scoped request returns 204 No Content. Verify with:

curl -s "http://connect-worker:8083/admin/loggers/org.apache.kafka.connect.runtime.WorkerSinkTask"

Worker-scoped changes are in-memory and do not survive a restart; cluster-scoped changes persist in the config topic. Use cluster scope for a deliberate, lasting change and worker scope for a one-off debug session on a single misbehaving worker — then revert to INFO to avoid drowning the log pipeline, since DEBUG on WorkerSinkTask logs per-batch detail.

Log Patterns Worth Alerting On

• Task threw an uncaught and unrecoverable exception (ERROR) — permanent task failure; the task moves to FAILED and will not retry on its own.
• CommitFailedException (ERROR) — the consumer left the group during an offset commit, usually because processing exceeded max.poll.interval.ms; transient but worth tracking if frequent.
• RetriableException (WARN) repeated > 5 times in 10 minutes — a persistent downstream problem the connector keeps retrying.
• TimeoutException during offset commits — broker or network issue.

Correlate spikes in these against offset-commit-failure-percentage and put-batch-avg-time-ms in dashboards.

Connector Lifecycle Management

Treat connector configurations as code: version-control the JSON, review changes in PRs, and apply through CI/CD. Manual creation via the REST API or a UI does not scale or audit.

Example: Debezium PostgreSQL Source

{
  "name": "pg-source-inventory",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "pg-primary.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/etc/connect/secrets.properties:pg-password}",
    "database.dbname": "inventory",
    "topic.prefix": "dbserver1",
    "table.include.list": "public.orders,public.customers",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_inventory",
    "publication.autocreate.mode": "filtered",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false"
  }
}

topic.prefix is the modern Debezium 2.x key; older database.server.name is deprecated. A PostgreSQL CDC source runs a single task (tasks.max: 1) because the logical replication slot is read sequentially.

Idempotent Deployment Script

PUT /connectors/{name}/config creates the connector if it does not exist and updates it otherwise, so a probe-then-branch is optional but keeps logs clear:

#!/bin/bash
set -euo pipefail

CONNECT_URL="http://connect-cluster:8083"
CONNECTOR_NAME="pg-source-inventory"
CONFIG_FILE="connectors/${CONNECTOR_NAME}.json"

# Apply just the "config" object via the idempotent PUT endpoint.
jq '.config' "${CONFIG_FILE}" | curl -s -X PUT \
  -H "Content-Type: application/json" \
  -d @- \
  "${CONNECT_URL}/connectors/${CONNECTOR_NAME}/config" | jq .

PUT /connectors/{name}/config expects the bare config object (the inner {...} of config), not the wrapper containing name/config, which is why the script extracts .config with jq.

Externalizing Secrets with Config Providers

Connect resolves ${provider:path:key} placeholders at task startup via pluggable ConfigProviders (KIP-297). The built-in FileConfigProvider needs no extra dependencies:

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

"database.password": "${file:/etc/connect/secrets.properties:pg-password}"

Apache Kafka ships FileConfigProvider and DirectoryConfigProvider; there is no built-in Vault provider. For HashiCorp Vault, install a community plugin such as jcustenborder/kafka-config-provider-vault — note the package is kafka.config.vault, not kafka.connect.vault:

config.providers=vault
config.providers.vault.class=com.github.jcustenborder.kafka.config.vault.VaultConfigProvider
config.providers.vault.param.vault.address=https://vault.internal:8200
config.providers.vault.param.vault.token=${file:/etc/connect/vault-token.properties:token}

Confluent also publishes a Vault provider; pick one implementation and follow its documented reference syntax, since the placeholder format differs between plugins.

Error Handling and Dead Letter Queues

By default a sink connector is intolerant of bad records: errors.tolerance defaults to none, so the first record that fails to convert, deserialize, or pass through a transform throws the task straight to FAILED. One malformed message — a null Avro payload, a schema the converter cannot read, a value an SMT rejects — then stops the entire partition. KIP-298 added per-record error tolerance and a dead letter queue (DLQ) so a poison record is routed aside instead of halting the pipeline:

# Sink connector config
errors.tolerance=all
errors.deadletterqueue.topic.name=connect-dlq-inventory
errors.deadletterqueue.topic.replication.factor=3
errors.deadletterqueue.context.headers.enable=true
errors.log.enable=true
errors.log.include.messages=false
errors.retry.timeout=0
errors.retry.delay.max.ms=60000

The defaults and behavior that bite operators:

The combination to understand: the DLQ and the retry/log settings apply only to converter and transform errors and (for sinks) errors the framework surfaces from put() — they do not catch logic errors inside a connector’s own delivery code, which still fail the task. The biggest silent-data-loss footgun is errors.tolerance=all with no DLQ topic set: records that fail conversion vanish with no trace. Always pair tolerance=all with a DLQ.

Detect the failure mode in metrics rather than waiting for downstream to notice missing data. Watch the DLQ topic’s own production rate — a non-zero rate() on records into connect-dlq-inventory means records are being diverted — and alert on it; a connector that is running while quietly DLQ-ing every record looks healthy on status alone.

Handling Rebalances and Task Failures

Connect distributes tasks across workers. Since version 2.3.0, distributed workers use incremental cooperative rebalancing by default (KIP-415): a worker joining or leaving only reassigns the affected connectors and tasks instead of stopping the world. This is built into the Connect group protocol — it is not the consumer partition.assignment.strategy/CooperativeStickyAssignor setting, which governs ordinary consumer groups, not Connect’s worker group. There is no config to opt into it; it is the default behavior.

Tuning Rebalance Behavior

# Worker config
group.id=connect-cluster
session.timeout.ms=30000
heartbeat.interval.ms=10000

# How long a leader waits before reassigning a departed worker's tasks.
# Lets a worker restart (e.g. a rolling upgrade) without churning assignments.
scheduled.rebalance.max.delay.ms=300000

scheduled.rebalance.max.delay.ms is a real Connect worker config (default 300000, i.e. 5 minutes) introduced with incremental cooperative rebalancing. Raise it to tolerate longer restarts; lower it to recover from a lost worker sooner at the cost of more reassignments.

Restarting Failed Connectors and Tasks

Connect does not automatically restart a task that has moved to FAILED. The key gotcha: POST /connectors/{name}/restart without query parameters restarts only the Connector instance, not its tasks — so a failed task stays failed. KIP-745 (Kafka 3.0) added includeTasks and onlyFailed to restart failed tasks in one call.

Before blindly restarting, read the failure. The status endpoint returns a trace field on any FAILED connector or task containing the exception stack trace — capturing it is what lets the escalation step decide whether a restart can possibly help:

#!/bin/bash
set -euo pipefail
CONNECT_URL="http://connect-cluster:8083"

# Find connectors with any FAILED connector or task instance, and log the trace.
curl -s "${CONNECT_URL}/connectors?expand=status" | jq -r '
  to_entries[]
  | select(
      .value.status.connector.state == "FAILED"
      or (.value.status.tasks[]?.state == "FAILED")
    )
  | .key as $name
  | ( [ .value.status.connector, .value.status.tasks[]? ]
      | map(select(.state == "FAILED"))[]
      | "\($name)\t\(.id // "connector")\t\(.trace | split("\n")[0])" )' \
  | tee /var/log/connect-failures.tsv \
  | cut -f1 | sort -u | while read -r connector; do
      echo "Restarting failed instances of: ${connector}"
      # includeTasks=true + onlyFailed=true restarts only the FAILED task/connector instances.
      curl -s -o /dev/null -w "%{http_code}\n" -X POST \
        "${CONNECT_URL}/connectors/${connector}/restart?includeTasks=true&onlyFailed=true"
    done

A successful restart returns 202 Accepted; restarting during an active rebalance returns 409 Conflict, so the loop should treat 409 as retryable. On Kafka < 3.0, restart each failed task individually instead:

curl -s -X POST "${CONNECT_URL}/connectors/${connector}/tasks/0/restart"

Escalation Policy

The first line of the captured trace usually classifies the failure, and the class determines whether a restart is futile:

Track restart count per connector. If a connector accumulates more than 3 restarts in an hour, stop auto-restarting and page on-call — repeated failures usually mean a poison record, a schema break, or a dead upstream, none of which a restart fixes. A Kubernetes CronJob running the script every few minutes is a common pattern; integrate the escalation with PagerDuty or Opsgenie.

Capacity Planning and Performance Tuning

Worker Sizing

Each task runs in its own thread; total in-flight concurrency is the sum of running tasks (bounded by each connector’s tasks.max) across the cluster. Scale out by adding workers — incremental cooperative rebalancing redistributes tasks with minimal disruption.

# Workers above 80% CPU
rate(process_cpu_seconds_total{job="kafka-connect"}[5m]) > 0.8

Two ceilings cap useful parallelism regardless of worker count. A sink connector cannot run more useful tasks than the source topic has partitions — extra tasks sit idle with zero assigned partitions. A source connector is capped by what the connector itself reports: a Debezium CDC source reads one replication slot sequentially, so tasks.max above 1 is wasted. Size tasks.max to min(target parallelism, partition count) for sinks, and to the connector’s documented maximum for sources.

Sink Connector Tuning

Client settings for a sink connector’s embedded consumer go in the connector config with the consumer.override. prefix (KIP-458, Kafka 2.3+), and require the worker to allow it via connector.client.config.override.policy=All (or Principal). The same keys without a prefix in the worker config (consumer.max.poll.records) set the cluster-wide default.

The max.poll.records/max.poll.interval.ms pair is the one to get right together: raising max.poll.records for throughput also raises the time to process one batch, and if that processing time can exceed max.poll.interval.ms the consumer is evicted mid-batch, the offset commit fails, and the task rebalances — often in a loop. Set max.poll.interval.ms from a measured worst-case batch time with margin, not the 5-minute default, whenever you raise max.poll.records past a few thousand.

Source Connector Tuning (Debezium)

These are Debezium connector properties; confirm exact defaults against your connector version, since they drift between releases.

Schema Registry Caching

When using Avro or Protobuf converters with Schema Registry, schema lookups add latency. Size the in-memory cache so steady-state operation never re-fetches:

key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081
key.converter.schema.registry.cache.capacity=1000
value.converter.schema.registry.cache.capacity=1000

The cache holds distinct schema IDs, so size it above the number of schema versions in flight, not the record rate — under the default capacity of 1000, a topic cycling through more than 1,000 schema versions thrashes and re-fetches on a cache miss, adding a Registry round-trip to the hot path. Incompatible schema changes surface as serialization failures that fail the task. Enforce compatibility checks in CI/CD before deploying connector or producer changes — see Schema Evolution and Compatibility.

Backup, Disaster Recovery, and Upgrades

Internal Topics

Distributed Connect persists all cluster state in three compacted internal topics:

A clarification worth internalizing: only source offsets live in connect-offsets. Sink connectors are ordinary consumers and commit their progress to __consumer_offsets under the connect-<connector> group, which is why sink recovery is governed by consumer offset retention, not these topics.

Back up connect-configs (the only one you cannot reconstruct from Git if you practice GitOps):

kafka-console-consumer --bootstrap-server kafka1:9092 \
  --topic connect-configs \
  --from-beginning \
  --timeout-ms 10000 \
  --property print.key=true \
  --property print.value=true \
  > connect-configs-backup.txt

Without --timeout-ms, kafka-console-consumer blocks indefinitely after draining the topic. The output is line-delimited key\tvalue, not valid JSON — restore by replaying, or simply re-apply connector configs from version control.

Disaster Recovery

1. Ensure the Kafka cluster is healthy and the three internal topics exist (same names, cleanup.policy=compact).
2. Re-apply connector configs from Git (preferred) or restore connect-configs/connect-offsets from backup.
3. Start workers with the original group.id.
4. Workers read configs and resume sources from connect-offsets; sinks resume from their consumer group offsets.

The recovery RPO differs by connector type and is easy to get wrong: a source connector resumes from connect-offsets, so restoring a stale copy of that topic re-reads — and re-emits — every change since the snapshot (at-least-once, with duplicates downstream). A sink resumes from __consumer_offsets; if those are lost or expired, recovery falls back to auto.offset.reset and can silently skip or replay data. Back up connect-offsets on the same cadence as connect-configs, and confirm __consumer_offsets retention (offsets.retention.minutes) outlasts your worst-case outage so a long incident does not expire a sink’s progress.

Rolling Upgrades

Connect supports rolling upgrades; scheduled.rebalance.max.delay.ms is what makes a one-worker-at-a-time restart cheap, since the leader holds the departed worker’s tasks rather than reshuffling immediately.

1. Stop one worker.
2. Upgrade binaries and connector plugins.
3. Restart it and wait for it to rejoin (rebalancing returns to false).
4. Watch connect-worker-rebalance-metrics rebalance-avg-time-ms and completed-rebalances-total for stability before moving on.
5. Repeat per worker.

Pre-Upgrade Validation

• Test in staging with a representative workload.
• The public Connect plugin API (SourceTask.poll(), SinkTask.put()) is stable, but connectors that reach into framework internals can break across versions.
• Check each connector’s release notes for compatibility with the target Kafka version.

Runbook

Maintain a runbook in the same repository as your connector configs:

• Topology: workers, brokers, Schema Registry endpoints.
• Connector inventory: name, class, tasks.max, owner.
• Recovery procedures with the exact, tested commands for restart, restore, and scale-out.

Automate provisioning with Terraform or Ansible so the documented state and the running state cannot drift.