Troubleshooting Kafka Connect Task Failures and Restarts

When operating Apache Kafka Connect in production, task failures are inevitable. Connectors and their tasks run as distributed processes that interact with external systems, and transient network blips, schema mismatches, resource exhaustion, or bad records can all push a task into the FAILED state. Knowing how to diagnose, recover from, and prevent these failures is a core competency for platform and SRE engineers managing the Schema Registry & Connect Ecosystem.

This guide focuses on the lifecycle of a failed task: the tools to inspect and restart it, the error-handling configuration that keeps transient problems from becoming permanent failures, and the monitoring that catches trouble early. It assumes you already understand how workers, connectors, and tasks relate. For a refresher on the components, see Kafka Connect Architecture and Deployment Modes.

Understanding Task States and the Failure Lifecycle

A connector or task reported through the REST API is in one of the following states: UNASSIGNED, RUNNING, PAUSED, FAILED, or RESTARTING. The RESTARTING state was introduced in Apache Kafka 3.0 (KIP-745); connector-level STOPPED was added in 3.5 (KIP-875). A task transitions to FAILED when it throws an exception that the Connect framework does not retry. Once a task fails, the connector as a whole keeps running if its other tasks are healthy, but the failed task stops producing or consuming records until an operator intervenes — Connect does not auto-restart a failed task.

Connect does have a built-in retry mechanism for sink connectors, governed by errors.retry.timeout and errors.retry.delay.max.ms. This applies to retriable errors raised in the converter, transformation, and put() stages of the sink pipeline. A RetriableException triggers an exponential backoff and re-attempt; a non-retriable exception, or one that persists past errors.retry.timeout, fails the task. By default errors.retry.timeout is 0, so no retries happen unless you opt in.

Distinguish a task failure from a connector failure. A connector failure — for example an invalid configuration that prevents the connector from starting at all — is a different problem. This guide concerns individual task failures within an otherwise healthy connector.

Inspecting Failed Tasks via the REST API

The Connect REST API is your primary diagnostic tool. All examples assume a distributed-mode cluster with workers on the default port 8083.

List all connector names:

curl -s http://localhost:8083/connectors | jq .

Get the status of one connector, including the state and trace of each task:

curl -s http://localhost:8083/connectors/source-postgres-inventory/status | jq .

The response includes a tasks array. A failed task shows "state": "FAILED" and a trace field with the full stack trace:

{
  "name": "source-postgres-inventory",
  "connector": {
    "state": "RUNNING",
    "worker_id": "10.0.1.5:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "FAILED",
      "worker_id": "10.0.1.5:8083",
      "trace": "org.apache.kafka.connect.errors.ConnectException: ..."
    }
  ],
  "type": "source"
}

To survey an entire cluster in a single call, use the expand parameter (added in Kafka 2.3, KIP-465), which returns each connector keyed by name with its full status inline:

curl -s "http://localhost:8083/connectors?expand=status" | jq .

You can also retrieve the connector configuration to check error-handling settings:

curl -s http://localhost:8083/connectors/source-postgres-inventory/config | jq .

For deeper diagnosis, read the worker logs. Connect logs task failures at ERROR level with the full exception chain. In distributed mode, check the worker that owns the failed task — identified by the worker_id field in the status response — since that is where the task and its log output live.

Common Failure Modes and Their Root Causes

Most task failures fall into a few categories. Recognizing the pattern speeds resolution.

Schema and serialization errors. When using Schema Registry, a task can fail if a record’s schema is incompatible with the registered version or if a schema ID is unknown. The trace typically mentions SerializationException (often a kafka.common.errors.SerializationException wrapping an Avro/Protobuf converter error). This is common during schema evolution when compatibility is not strictly enforced.

External-system connectivity. Source and sink tasks that lose connectivity to the database, REST endpoint, or object store usually surface a ConnectException wrapping a root cause such as SQLException or SocketTimeoutException. Transient blips may clear on retry; prolonged outages require a manual restart once the dependency recovers.

Resource exhaustion. Tasks can fail with OutOfMemoryError when the heap is too small for the connector’s working set, or with timeouts when the external system is overloaded. Sink tasks that batch large writes may exceed the broker’s message.max.bytes (or the topic’s max.message.bytes) or the producer’s max.request.size, producing a RecordTooLargeException.

Conversion and transformation errors. A Single Message Transform (SMT) can throw DataException when an expected field is missing or has an unexpected type. A sink connector can likewise fail on a key or value it cannot map to the target system.

Reassignment after rebalance. In distributed mode, tasks pass through UNASSIGNED during worker rebalances. A task that fails immediately on reassignment often points to a stale configuration or a worker that cannot reach the external system from its network location.

Restarting Failed Tasks Safely

The narrowest recovery is to restart the single failed task. This reinitializes the task and resumes from the last committed offset (source offsets for source connectors, consumer offsets for sink connectors):

curl -s -X POST http://localhost:8083/connectors/source-postgres-inventory/tasks/0/restart

A successful task restart returns HTTP 204 (No Content), so there is no JSON body to pipe through jq.

The bare connector restart endpoint is narrower than it looks. By default POST /connectors/{name}/restart restarts only the Connector instance, not its tasks. Before KIP-745 (Kafka 3.0), recovering several failed tasks meant one REST call per task. Since 3.0 you can restart the connector and all of its failed tasks in a single call:

curl -s -X POST "http://localhost:8083/connectors/source-postgres-inventory/restart?includeTasks=true&onlyFailed=true"

includeTasks=true tells Connect to restart task instances as well as the connector; onlyFailed=true limits the restart to instances currently in FAILED status, leaving healthy tasks untouched. This is the recommended recovery for a connector with multiple failed tasks. The endpoint returns HTTP 202 (Accepted) with the list of instances being restarted, or 409 (Conflict) if a rebalance is in progress.

Before restarting, capture the task trace and relevant log excerpts. If the failure recurs you will need that data for root-cause analysis. A runbook that records task status before and after the restart gives you an audit trail.

Configuring Error Handling and Dead Letter Queues

To keep individual bad records or transient errors from killing a task, tune the framework error-handling configuration. These properties — introduced in KIP-298 — apply to sink connectors (they govern the converter, transformation, and put() stages). The dead letter queue in particular is sink-only:

errors.tolerance=all
errors.log.enable=true
errors.log.include.messages=true
errors.deadletterqueue.topic.name=dlq.my-connector
errors.deadletterqueue.topic.replication.factor=3
errors.deadletterqueue.context.headers.enable=true
errors.retry.timeout=300000
errors.retry.delay.max.ms=60000

errors.tolerance defaults to none, which fails the task on the first error. Setting errors.tolerance=all skips records that error during conversion or transformation rather than failing the task. Pair it with a dead letter queue: skipped records are written to the configured DLQ topic for later inspection and replay. Set errors.deadletterqueue.context.headers.enable=true so each DLQ record carries headers describing the original topic, partition, offset, and the exception — without them, triaging the DLQ is guesswork. The DLQ topic is not auto-created with the right replication unless you set errors.deadletterqueue.topic.replication.factor (default 3) appropriately for your cluster; on a single-broker dev cluster you must lower it or the connector cannot create the topic.

Two defaults catch operators off guard: errors.log.enable and errors.log.include.messages are both false, so failed records are not logged unless you turn them on, and errors.retry.timeout is 0, so retriable errors are not retried by default.

Source connectors do not support the dead letter queue. Their error handling is more limited because the connector itself controls source-offset commits; consult each source connector’s own documentation for what it supports. See the Apache Kafka Connect documentation for the canonical reference on these properties.

Monitoring and Alerting on Task Health

Proactive monitoring catches failures before they cause lag or data loss. Connect exposes metrics over JMX and exposes status over the REST API. Wire both into your observability stack — and understand a sharp edge in the task-level metrics.

Relevant MBeans:

  • kafka.connect:type=connector-task-metrics,connector="{name}",task="{id}" — per-task metrics including status (unassigned/running/paused/failed/destroyed), running-ratio, pause-ratio, and offset-commit timings.
  • kafka.connect:type=connect-worker-metrics,connector="{name}" — per-connector task counts on this worker: connector-total-task-count, connector-running-task-count, connector-paused-task-count, connector-failed-task-count, connector-unassigned-task-count, and connector-destroyed-task-count (KIP-475).
  • kafka.connect:type=connect-worker-metrics — worker-wide task-count and connector-count.

Gotcha: the per-task connector-task-metrics MBean — and therefore its status attribute — is not registered while a task is in the FAILED state (KAFKA-9066). Alerting on status="failed" is unreliable because the bean disappears exactly when you need it. Alert instead on the worker metric connector-failed-task-count, which is always present and counts failed tasks directly.

A REST-based health check sidesteps the JMX limitation entirely. The status endpoint always reports failed tasks:

#!/bin/bash
FAILED=$(curl -s "http://localhost:8083/connectors?expand=status" | \
  jq -r '[ .[].status | select(.tasks[]?.state == "FAILED") | .name ] | unique | @csv')
if [ -n "$FAILED" ]; then
  echo "ALERT: connectors with failed tasks: $FAILED"
  exit 1
fi

(The expand=status response is a map keyed by connector name; each value has a .status object holding the same shape as GET /connectors/{name}/status, which is why the filter reaches through .[].status.)

For Prometheus, scrape Connect with the JMX exporter and alert on the worker metric, for example kafka_connect_connect_worker_metrics_connector_failed_task_count > 0 (the exact series name depends on your exporter’s naming rules). Do not rely on a status="failed" series from the task metric, for the KAFKA-9066 reason above.

Finally, monitor consumer lag on the DLQ topic. A steadily growing DLQ signals a systemic data-quality problem that needs attention even though the connector is happily tolerating errors.

Preventing Task Failures Through Operational Discipline

Many failures are preventable.

Schema-evolution discipline. Enforce a Schema Registry compatibility mode (BACKWARD, FORWARD, or FULL) that matches your data contracts, and test schema changes in staging before promoting them. The Confluent documentation on schema evolution and compatibility details the rules.

Resource sizing. Size worker heap for the number and type of connectors, and watch GC pauses, adjusting via KAFKA_HEAP_OPTS. For sink connectors, tune consumer.override.max.poll.records together with the worker offset.flush.timeout.ms so slow processing does not trip the consumer-group max.poll.interval.ms and trigger a rebalance. (Per-connector consumer overrides require connector.client.config.override.policy=All on the worker, which is the default since Kafka 3.0.)

Validate before deploying. Test a configuration with PUT /connector-plugins/{connector-type}/config/validate to catch missing required fields, invalid enum values, and type mismatches before they fail a live task:

curl -s -X PUT -H "Content-Type: application/json" \
  --data @my-connector.json \
  http://localhost:8083/connector-plugins/io.confluent.connect.jdbc.JdbcSinkConnector/config/validate | jq .

Graceful maintenance. To take a connector out of service cleanly — for example during worker maintenance — PUT /connectors/{name}/pause stops its tasks while preserving its configuration and committed offsets, and PUT .../resume brings it back. From Kafka 3.5, PUT .../stop shuts the tasks down and releases their resources while keeping offsets. Reserve DELETE /connectors/{name} for genuinely removing a connector: it halts all tasks and deletes the configuration, so it is not the right tool for a temporary pause. For rolling worker restarts, restart one worker at a time and wait for task reassignment to stabilize before moving on.

Treat configuration as code. Store connector configs in version control and apply them through a pipeline that validates and deploys them. This prevents drift and makes rollback a one-command operation when a change introduces failures.

Combine robust sink error handling, monitoring built on metrics that survive failure, and disciplined operations, and you minimize both the frequency and the blast radius of task failures. When one does occur, the diagnostic and recovery steps above restore service quickly and safely.

Back to Kafka Connect Architecture And Deployment Modes