Important

You are viewing documentation for an older version of Confluent Platform. For the latest, click here.

Monitoring Kafka Connect and Connectors

You can manage and monitor Connect, connectors, and clients using JMX and the REST interface.

Using the REST interface

Kafka Connect’s REST API enables administration of the cluster. This includes APIs to view the configuration of connectors and the status of their tasks, as well as to alter their current behavior (for example, changing configuration and restarting tasks).

Once a Kafka Connect cluster is up and running, you can monitor and modify it. In this section, we go over a few common management tasks done via the REST API.

Since Kafka Connect is intended to be run as a service, it also supports a REST API for managing connectors. By default this service runs on port 8083. When executed in distributed mode, the REST API will be the primary interface to the cluster. You can make requests to any cluster member; the REST API automatically forwards requests if required.

Although you can use the standalone mode just by submitting a connector on the command line, it also runs the REST interface. This is useful for getting status information, adding and removing connectors without stopping the process, and more.

Connector and task status

You can use the REST API to view the current status of a connector and its tasks, including the ID of the worker to which each was assigned.

Connectors and their tasks publish status updates to a shared topic (configured with status.storage.topic) which all workers in the cluster monitor. Because the workers consume this topic asynchronously, there is typically a short delay before a state change is visible through the status API. The following states are possible for a connector or one of its tasks:

  • UNASSIGNED: The connector/task has not yet been assigned to a worker.
  • RUNNING: The connector/task is running.
  • PAUSED: The connector/task has been administratively paused.
  • FAILED: The connector/task has failed (usually by raising an exception, which is reported in the status output).

In most cases, connector and task states will match, though they may be different for short periods of time when changes are occurring or if tasks have failed. For example, when a connector is first started, there may be a noticeable delay before the connector and its tasks have all transitioned to the RUNNING state. States will also diverge when tasks fail since Connect does not automatically restart failed tasks.

It’s sometimes useful to temporarily stop the message processing of a connector. For example, if the remote system is undergoing maintenance, it would be preferable for source connectors to stop polling it for new data instead of filling logs with exception spam. For this use case, Connect offers a pause/resume API. While a source connector is paused, Connect will stop polling it for additional records. While a sink connector is paused, Connect will stop pushing new messages to it. The pause state is persistent, so even if you restart the cluster, the connector will not begin message processing again until the task has been resumed. Note that there may be a delay before all of a connector’s tasks have transitioned to the PAUSED state since it may take time for them to finish whatever processing they were in the middle of when being paused. Additionally, failed tasks will not transition to the PAUSED state until they have been restarted.

Common REST examples

Below are a few common activities that are done over the REST API. These examples are shown using a worker running on localhost with default configurations and a connector named s3-connector. The utility jq is used in the examples to format the response, but this is not required.

Tip

The examples show basic commands.

  • Get worker cluster ID, version, and git source code commit ID:

    curl localhost:8083/ | jq

    Example output:

    {
  "version": "5.5.0-ce",
  "commit": "2bedb2c6980ba7a8",
  "kafka_cluster_id": "wxOOb2vrTPCLJl28PI18KA"
}

  • List the connector plugins available on a worker:

    curl localhost:8083/connector-plugins | jq

    Example output:

    [
  {
    "class": "io.confluent.connect.activemq.ActiveMQSourceConnector",
    "type": "source",
    "version": "5.5.0"
  },
  {
    "class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "type": "sink",
    "version": "5.5.0"
  },
  {
    "class": "io.confluent.connect.ibm.mq.IbmMQSourceConnector",
    "type": "source",
    "version": "5.5.0"
  },
  {
    "class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "type": "sink",
    "version": "5.5.0"
  },
  {
    "class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "type": "source",
    "version": "5.5.0"
  },
  {
    "class": "io.confluent.connect.jms.JmsSourceConnector",
    "type": "source",
    "version": "5.5.0"
  },

  ... omitted

  • List active connectors on a worker:

    curl localhost:8083/connectors

    Example output:

    ["s3-connector"]

  • Restart a connector (there is no output if the command is successful):

    curl -X POST localhost:8083/connectors/s3-connector/restart

    Note

    Restarting a connector does not restart the tasks.

  • Get connector tasks:

    curl localhost:8083/connectors/s3-connector/tasks | jq

    Example output:

    [
  {
    "id": {
      "connector": "s3-connector",
      "task": 0
    },
    "config": {
      "connector.class": "io.confluent.connect.s3.S3SinkConnector",
      "format.class": "io.confluent.connect.s3.format.bytearray.ByteArrayFormat",
      "task.class": "io.confluent.connect.s3.S3SinkTask",
      "flush.size": "1000",
      "topics": "passengers",
      "name": "s3-connector",
      "aws.access.key.id": "omitted",
      "rotate.interval.ms": "-3",
      "aws.secret.access.key": "omitted",
      "storage.class": "io.confluent.connect.s3.storage.S3Storage",
      "s3.bucket.name": "s3-test-bucket"
    }
  }
]

  • Restart a task (there is no output if the command is successful):

    curl -X POST localhost:8083/connectors/s3-connector/tasks/0/restart

  • Pause a connector (there is no output if the command is successful):

    curl -X PUT localhost:8083/connectors/s3-connector/pause

    Tip

    This command is useful if the system the connector interacts with is required to be taken out of service temporarily for maintenance.

  • Resume a paused connector (there is no output if the command is successful):

    curl -X PUT localhost:8083/connectors/s3-connector/resume

  • Update the connector configuration:

    curl -X PUT -H "Content-Type: application/json" --data '{"connector.class":"io.confluent.connect.s3.S3SinkConnector","format.class":"io.confluent.connect.s3.format.bytearray.ByteArrayFormat","flush.size":"1000","s3.bucket.name":"s3-test-bucket","storage.class":"io.confluent.connect.s3.storage.S3Storage","tasks.max":"2","topics":"passengers","name":"s3-connector"}' localhost:8083/connectors/s3-connector/config | jq

    The example command updates tasks from 1 to 2:

    {
  "name": "s3-connector",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "format.class": "io.confluent.connect.s3.format.bytearray.ByteArrayFormat",
    "flush.size": "1000",
    "s3.bucket.name": "s3-test-bucket",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "tasks.max": "2",
    "topics": "passengers",
    "name": "s3-connector"
  },
  "tasks": [
    {
      "connector": "s3-connector",
      "task": 0
    },
    {
      "connector": "s3-connector",
      "task": 1
    }
  ],
  "type": "sink"
}

  • Get connector status:

    curl localhost:8083/connectors/s3-connector/status | jq

    Example output:

    {
   "name": "s3-connector",
   "connector": {
      "state": "RUNNING",
      "worker_id": "192.168.86.66:8083"
   },
   "tasks": [
   {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "192.168.86.66:8083"
   },
   {
      "id": 1,
      "state": "RUNNING",
      "worker_id": "192.168.86.66:8083"
   }
   ],
   "type": "sink"
}

  • Get the connector configuration, tasks, and type of connector:

    curl localhost:8083/connectors/s3-connector | jq

    Example output:

    {
   "name": "s3-connector",
   "config": {
      "connector.class": "io.confluent.connect.s3.S3SinkConnector",
      "format.class": "io.confluent.connect.s3.format.bytearray.ByteArrayFormat",
      "flush.size": "1000",
      "tasks.max": "2",
      "topics": "passengers",
      "name": "s3-connector",
      "storage.class": "io.confluent.connect.s3.storage.S3Storage",
      "s3.bucket.name": "s3-test-bucket"
   },
   "tasks": [
   {
      "connector": "s3-connector",
      "task": 0
   },
   {
      "connector": "s3-connector",
      "task": 1
   }
   ],
   "type": "sink"
}

    To get the connector configuration only, use the following command:

    curl localhost:8083/connectors/s3-connector/config | jq

  • Delete a connector (there is no output if the command is successful):

    curl -X DELETE localhost:8083/connectors/s3-connector

Using JMX to monitor Connect

Connect reports a variety of metrics through Java Management Extensions (JMX). Connect can be configured to report stats using additional pluggable stats reporters using the metric.reporters configuration option. The following sections provide MBeans, metrics, and descriptions.

Connector metrics

MBean: kafka.connect:type=connector-metrics,connector=”{([-.w]+)

Metric Explanation
connector-type Connector type: source or sink
connector-class Connector class name
connector-version Connector class version (as reported by the connector)
status Connector status: running, paused, or stopped

Common task metrics

MBean: kafka.connect:type=connector-task-metrics,connector=”{connector}”,task=”{task}”

Metric Explanation
status Current task status: unassigned, running, paused, failed, or destroyed
pause-ratio Fraction of time the task has spent in a paused state
running-ratio Fraction of time the task has spent in the running state
offset-commit-success-percentage Average percentage of the task’s offset commit attempts that succeeded
offset-commit-failure-percentage Average percentage of the task’s offset commit attempts that failed or had an error
offset-commit-max-time-ms Maximum time in milliseconds taken by the task to commit offsets
offset-commit-avg-time-ms Average time in milliseconds taken by the task to commit offsets
batch-size-max Maximum size of the batches processed by the connector
batch-size-avg Average size of the batches processed by the connector

Worker metrics

MBean: kafka.connect:type=connect-worker-metrics

Metric Explanation
task-count Number of tasks that have run in this worker
connector-count The number of connectors that have run in this worker
connector-startup-attempts-total Total number of connector startups that this worker has attempted
connector-startup-success-total Total number of connector starts that succeeded
connector-startup-success-percentage Average percentage of the worker’s connector starts that succeeded
connector-startup-failure-total Total number of connector starts that failed
connector-startup-failure-percentage Average percentage of the worker’s connectors starts that failed
task-startup-attempts-total Total number of task startups that the worker has attempted
task-startup-success-total Total number of task starts that succeeded
task-startup-success-percentage Average percentage of the worker’s task starts that succeeded
task-startup-failure-total Total number of task starts that failed
task-startup-failure-percentage Average percentage of the worker’s task starts that failed

Worker rebalance metrics

MBean: kafka.connect:type=connect-worker-metrics

Metric Explanation
leader-name Name of the group leader
epoch Epoch or generation number of the worker
completed-rebalances-total Total number of rebalances completed by the worker
rebalancing Whether the worker is currently rebalancing
rebalance-max-time-ms Maximum time the worker spent rebalancing (in milliseconds)
rebalance-avg-time-ms Average time the worker spent rebalancing (in milliseconds)
time-since-last-rebalance-ms Time since the most recent worker rebalance (in milliseconds)

Source task metrics

MBean: kafka.connect:type=source-task-metrics,connector=([-.w]+),task=([d]+)

Metric Explanation
source-record-write-total Number of records output from the transformations and written to Kafka for the task belonging to the named source connector in the worker (since the task was last restarted)
source-record-write-rate After transformations are applied, this is the average per-second number of records output from the transformations and written to Kafka for the task belonging to the named source connector in the worker (excludes any records filtered out by the transformations)
source-record-poll-total Before transformations are applied, this is the number of records produced or polled by the task belonging to the named source connector in the worker (since the task was last restarted)
source-record-poll-rate Before transformations are applied, this is the average per-second number of records produced or polled by the task belonging to the named source connector in the worker
source-record-active-count-max Maximum number of records produced by the task but not yet completely written to Kafka
source-record-active-count-avg Average number of records produced by the task but not yet completely written to Kafka
source-record-active-count Most recent number of records produced by the task but not yet completely written to Kafka
poll-batch-max-time-ms Maximum time in milliseconds taken by this task to poll for a batch of source records
poll-batch-avg-time-ms Average time in milliseconds taken by this task to poll for a batch of source records

Sink task metrics

MBean: kafka.connect:type=sink-task-metrics,connector=([-.w]+),task=([d]+)

Metric Explanation
sink-record-read-rate Before transformations are applied, this is the average per-second number of records read from Kafka for the task belonging to the named sink connector in the worker
sink-record-read-total Before transformations are applied, this is the total number of records produced or polled by the task belonging to the named source connector in the worker (since the task was last restarted)
sink-record-send-rate After transformations are applied, this is the average per-second number of records output from the transformations and sent to the task belonging to the named sink connector in the worker (excludes any records filtered out by the transformations)
sink-record-send-total Total number of records output from the transformations and sent to the task belonging to the named source connector in the worker (since the task was last restarted)
sink-record-active-count Most recent number of records read from Kafka but not yet completely committed, flushed, or acknowledged by the sink task
sink-record-active-count-max Maximum number of records read from Kafka, but that have not yet completely been committed, flushed, or acknowledged by the sink task
sink-record-active-count-avg Average number of records read from Kafka, but that have not yet completely been committed, flushed, or acknowledged by the sink task
partition-count Number of topic partitions assigned to the task and which belong to the named sink connector in the worker
offset-commit-seq-no Current sequence number for offset commits
offset-commit-completion-rate Average per-second number of offset commit completions that have completed successfully
offset-commit-completion-total Total number of offset commit completions that were completed successfully
offset-commit-skip-rate Average per-second number of offset commit completions that were received too late and skipped, or ignored
offset-commit-skip-total Total number of offset commit completions that were received too late and skipped, or ignored
put-batch-max-time-ms Maximum time in milliseconds taken by this task to put a batch of sink records

Client metrics

MBean: kafka.connect:type=connect-metrics,client-id=

Metric Explanation
connection-close-rate Connections closed per second in the window
connection-count Current number of active connections
connection-creation-rate New connections established per second in the window
failed-authentication-rate Connections that failed authentication
incoming-byte-rate Bytes per second read off all sockets
io-ratio Fraction of time the I/O thread spent doing I/O
io-time-ns-avg Average length of time for I/O per select call in nanoseconds
io-wait-ratio Fraction of time the I/O thread spent waiting
io-wait-time-ns-avg Average length of time the I/O thread spent waiting for a socket ready for reads or writes in nanoseconds
network-io-rate Average number of network operations (reads or writes) on all connections per second
outgoing-byte-rate Average number of outgoing bytes sent per second to all servers
request-rate Average number of requests sent per second
request-size-avg Average size of all requests in the window
request-size-max Maximum size of any request sent in the window
response-rate Responses received and sent per second
select-rate Number of times the I/O layer checked for new I/O to perform per second
successful-authentication-rate Connections that were successfully authenticated using SASL or SSL