Kafka Connect REST API cheat sheet: status, restarting failed tasks, pause/stop and resetting offsets

#Before you start

Kafka Connect in distributed mode is managed through its REST API, by default on port 8083 of every worker. Any worker accepts the request and forwards writes to the leader, so you can point at one worker or at a load balancer.

export CONNECT=http://connect-worker-1:8083
# Add -u user:pass or --cert/--key if the REST listener is secured
curl -s $CONNECT/ | jq        # worker version, commit and Kafka cluster id

All examples use curl and jq. Connector names with spaces or slashes must be URL-encoded.

Warning

Some endpoints only exist in newer releases. Restarting only failed tasks needs Kafka 3.0 or later. stop needs 3.5 or later. Altering and resetting offsets needs 3.6 or later. If an endpoint returns 404 on a connector you know exists, check the worker version from GET / first.

#Health at a glance

#List connectors with their status

curl -s "$CONNECT/connectors?expand=status" | jq -r '
  to_entries[] | .key as $n | .value.status |
  "\($n)\t\(.connector.state)\t" + ([.tasks[] | "\(.id):\(.state)"] | join(" "))'

This prints one line per connector, for example orders-sink RUNNING 0:RUNNING 1:FAILED. A connector can report RUNNING while one of its tasks is FAILED. Always check task states too.

#Only the broken ones

curl -s "$CONNECT/connectors?expand=status" | jq -r '
  to_entries[] | select(
    .value.status.connector.state != "RUNNING" or
    any(.value.status.tasks[]; .state != "RUNNING"))
  | .key'

#Read the failure reason

curl -s $CONNECT/connectors/orders-sink/status | jq -r '.tasks[] | select(.state=="FAILED") | "task \(.id) on \(.worker_id)\n\(.trace)"' | head -40

The trace field holds the Java stack trace. Look for the first Caused by: line, which usually names the real problem.

#Common failure causes

#Restart

#Restart only failed tasks (Kafka 3.0+)

curl -s -X POST "$CONNECT/connectors/orders-sink/restart?includeTasks=true&onlyFailed=true" | jq

This returns 202 Accepted with the restart plan. It is the safest restart: healthy tasks keep running.

#Restart one task

curl -s -X POST $CONNECT/connectors/orders-sink/tasks/1/restart

#Restart the connector instance only

curl -s -X POST $CONNECT/connectors/orders-sink/restart

Without query parameters, only the connector instance restarts, not its tasks. This catches people out: the failed tasks stay failed.

#Restart every failed task across the cluster

for c in $(curl -s "$CONNECT/connectors?expand=status" | jq -r '
  to_entries[] | select(any(.value.status.tasks[]; .state=="FAILED")) | .key'); do
  echo "restarting $c"
  curl -s -o /dev/null -w "%{http_code}\n" -X POST \
    "$CONNECT/connectors/$c/restart?includeTasks=true&onlyFailed=true"
done

Tip

If a task fails again within seconds, restarting in a loop will not help. Read the trace and fix the cause, such as an expired credential, a missing ACL or a poison record.

#Pause, resume and stop

curl -s -X PUT $CONNECT/connectors/orders-sink/pause    # tasks stay assigned, stop processing
curl -s -X PUT $CONNECT/connectors/orders-sink/resume
curl -s -X PUT $CONNECT/connectors/orders-sink/stop     # 3.5+: tasks shut down, config kept

Use pause for short holds such as target maintenance. Use stop before any offset change, and when you want to free worker resources without deleting the connector.

#Config changes

# Current config
curl -s $CONNECT/connectors/orders-sink/config | jq

# Validate before applying (plugin class name in the path)
curl -s -X PUT -H "Content-Type: application/json" \
  --data @orders-sink.json \
  $CONNECT/connector-plugins/io.confluent.connect.jdbc.JdbcSinkConnector/config/validate \
  | jq '.error_count, [.configs[] | select(.value.errors | length > 0) | {name: .value.name, errors: .value.errors}]'

# Create or update (idempotent; body is the flat config map)
curl -s -X PUT -H "Content-Type: application/json" \
  --data @orders-sink.json $CONNECT/connectors/orders-sink/config | jq

PUT .../config creates the connector if it does not exist and replaces the whole config if it does. Keep connector configs in Git and apply them with PUT so changes are repeatable. Settings left out of the body go back to their defaults.

# Installed plugins and their versions
curl -s $CONNECT/connector-plugins | jq -r '.[] | "\(.class)\t\(.version)"'

# Topics a connector has used (since its last topics reset)
curl -s $CONNECT/connectors/orders-sink/topics | jq

#Change log levels without a restart

# Current non-default levels
curl -s $CONNECT/admin/loggers | jq

# Turn on DEBUG for one connector package on this worker
curl -s -X PUT -H "Content-Type: application/json" \
  -d '{"level":"DEBUG"}' $CONNECT/admin/loggers/io.confluent.connect.jdbc

# Set it back
curl -s -X PUT -H "Content-Type: application/json" \
  -d '{"level":"INFO"}' $CONNECT/admin/loggers/io.confluent.connect.jdbc

Log level changes are not saved and are lost when the worker restarts. In older releases they apply only to the worker you call. Newer releases accept ?scope=cluster. Check your version's documentation before relying on it.

#Offsets (Kafka 3.6+)

#Read offsets

curl -s $CONNECT/connectors/orders-sink/offsets | jq

For sink connectors, each entry has partition: {kafka_topic, kafka_partition} and offset: {kafka_offset}. For source connectors, the keys are defined by the connector plugin. For example, a JDBC source or Debezium stores its own position keys. Read the current offsets before you change them so you have the exact shape.

Caution

Changing offsets causes reprocessing or skipped data. Save the current offsets to a file first, and confirm with the data owner whether the target can handle duplicates.

#Move a sink connector to specific offsets

curl -s $CONNECT/connectors/orders-sink/offsets > orders-sink-offsets-backup.json
curl -s -X PUT $CONNECT/connectors/orders-sink/stop
curl -s $CONNECT/connectors/orders-sink/status | jq -r .connector.state   # must be STOPPED

curl -s -X PATCH -H "Content-Type: application/json" $CONNECT/connectors/orders-sink/offsets -d '{
  "offsets": [
    { "partition": { "kafka_topic": "orders", "kafka_partition": 0 },
      "offset":    { "kafka_offset": 1200000 } }
  ]
}' | jq

curl -s -X PUT $CONNECT/connectors/orders-sink/resume

PATCH only adds or changes the partitions you list. Other partitions keep their offsets. Setting an offset to null clears it for that partition.

#Reset all offsets

curl -s -X PUT $CONNECT/connectors/orders-sink/stop
curl -s -X DELETE $CONNECT/connectors/orders-sink/offsets | jq
curl -s -X PUT $CONNECT/connectors/orders-sink/resume

For a sink connector, a reset deletes its consumer group. Where it starts reading after the reset depends on the consumer auto.offset.reset setting. For a source connector, the connector starts as if it is new, which for CDC connectors may mean a fresh snapshot.

#Delete

curl -s -X DELETE $CONNECT/connectors/orders-sink

Deleting a connector removes its config but not its committed offsets. If you recreate a connector with the same name, it continues from where it left off. To start clean, reset the offsets before deleting it, or use a new name.

#Prevent it next time

• Alert on any task not in RUNNING, not just the connector state. Poll ?expand=status or use the worker's JMX task metrics.
• Set errors.tolerance, a dead-letter queue and errors.log.enable on sink connectors so a single bad record does not fail the task.
• Keep connector configs in Git and apply them with PUT .../config. Run config/validate in CI.
• Save each connector's offsets on a schedule (GET .../offsets) so you have a known-good position to roll back to.
• Put a load balancer in front of the workers and secure the REST listener. Anyone who can reach it can delete connectors or rewrite offsets.