Managing Connect with the REST API

Summary

The Connect REST API provides complete lifecycle management for connectors. You can create, read, update, delete, pause, resume, and restart connectors—all via HTTP.

All these operations go through the REST API exposed by every worker. You can send requests to any worker—they all share the same state through Kafka topics.

POST: Creating Connectors

Let's start with creating a connector. This is the most common operation.

Basic Syntax

1curl -X POST http://localhost:8083/connectors \
2  -H "Content-Type: application/json" \
3  -d '{
4    "name": "my-connector",
5    "config": {
6      "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
7      "tasks.max": "1",
8      "file": "/tmp/input.txt",
9      "topic": "file-events"
10    }
11  }'

Example: JDBC Source Connector

1curl -X POST http://localhost:8083/connectors \
2  -H "Content-Type: application/json" \
3  -d '{
4    "name": "mysql-source",
5    "config": {
6      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
7      "tasks.max": "3",
8      "connection.url": "jdbc:mysql://localhost:3306/mydb",
9      "connection.user": "kafka",
10      "connection.password": "secret",
11      "mode": "incrementing",
12      "incrementing.column.name": "id",
13      "topic.prefix": "mysql-",
14      "table.whitelist": "users,orders,products"
15    }
16  }'

Response

1{
2  "name": "mysql-source",
3  "config": {
4    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
5    "tasks.max": "3",
6    "connection.url": "jdbc:mysql://localhost:3306/mydb",
7    "mode": "incrementing",
8    "incrementing.column.name": "id",
9    "topic.prefix": "mysql-"
10  },
11  "tasks": [],
12  "type": "source"
13}

The response confirms the connector was created. The tasks array is initially empty because tasks are created asynchronously after the connector is configured.

GET: Querying Connectors

GET operations let you inspect connector state, configuration, and status.

List All Connectors

1curl http://localhost:8083/connectors

Response:

1["mysql-source", "elasticsearch-sink", "s3-sink"]

Get Connector Configuration

1curl http://localhost:8083/connectors/mysql-source

Get Connector Status

1curl http://localhost:8083/connectors/mysql-source/status

Response shows detailed status:

1{
2  "name": "mysql-source",
3  "connector": {
4    "state": "RUNNING",
5    "worker_id": "worker1:8083"
6  },
7  "tasks": [
8    {
9      "id": 0,
10      "state": "RUNNING",
11      "worker_id": "worker1:8083"
12    },
13    {
14      "id": 1,
15      "state": "RUNNING",
16      "worker_id": "worker2:8083"
17    },
18    {
19      "id": 2,
20      "state": "FAILED",
21      "worker_id": "worker3:8083",
22      "trace": "org.apache.kafka.connect.errors.ConnectException: ..."
23    }
24  ],
25  "type": "source"
26}

The status endpoint is crucial for monitoring. It shows whether the connector and each task are running, paused, or failed. If a task failed, the trace field provides the error message.

Here's how status monitoring works in practice:

Monitoring tools periodically query connector status and can automatically restart failed tasks or alert operations teams.

PUT: Updating Connectors

To update a connector's configuration, use PUT to the config endpoint:

1curl -X PUT http://localhost:8083/connectors/mysql-source/config \
2  -H "Content-Type: application/json" \
3  -d '{
4    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
5    "tasks.max": "5",
6    "connection.url": "jdbc:mysql://localhost:3306/mydb",
7    ...
8  }'

This replaces the entire configuration. Connect will restart the connector with the new settings. Be careful—this causes a brief interruption in data flow.

DELETE: Removing Connectors

To delete a connector and stop all its tasks:

1curl -X DELETE http://localhost:8083/connectors/mysql-source

This permanently removes the connector and all its tasks. The configuration is deleted from the config topic. However, any data already produced to Kafka remains—DELETE only removes the connector, not the data.

Pause and Resume

Sometimes you want to temporarily stop a connector without deleting it.

Pause a Connector

1curl -X PUT http://localhost:8083/connectors/mysql-source/pause

Resume a Connector

1curl -X PUT http://localhost:8083/connectors/mysql-source/resume

Pausing stops data movement but preserves configuration and offsets. When you resume, the connector picks up exactly where it left off. This is useful for maintenance windows or temporary slowdowns.

Restart Operations

When a connector or task fails, you can restart it:

Restart Connector

1curl -X POST http://localhost:8083/connectors/mysql-source/restart

Restart Specific Task

1curl -X POST http://localhost:8083/connectors/mysql-source/tasks/2/restart

Restarting is useful for transient errors like network timeouts or temporary database unavailability. The connector or task will retry from its last saved offset.

Advanced Operations

Here are some more advanced REST API operations:

Get Task Status

1curl http://localhost:8083/connectors/mysql-source/tasks/0/status

Get Connector Topics

1curl http://localhost:8083/connectors/mysql-source/topics

Shows which topics the connector is using.

Validate Configuration Before Deployment

1curl -X PUT http://localhost:8083/connector-plugins/JdbcSourceConnector/config/validate \
2  -H "Content-Type: application/json" \
3  -d @mysql-source-config.json

Returns validation errors without actually creating the connector.

List Available Connector Plugins

1curl http://localhost:8083/connector-plugins

Let's see the full lifecycle using REST API operations:

Practical Automation Example

Here's a bash script that automates connector deployment:

1#!/bin/bash
2CONNECT_HOST="http://localhost:8083"
3CONNECTOR_NAME="mysql-source"
4CONFIG_FILE="mysql-source-config.json"
5
6# Check if connector exists
7if curl -s "$CONNECT_HOST/connectors/$CONNECTOR_NAME" > /dev/null 2>&1; then
8  echo "Connector exists, updating..."
9  curl -X PUT "$CONNECT_HOST/connectors/$CONNECTOR_NAME/config" \
10    -H "Content-Type: application/json" \
11    -d @$CONFIG_FILE
12else
13  echo "Creating new connector..."
14  curl -X POST "$CONNECT_HOST/connectors" \
15    -H "Content-Type: application/json" \
16    -d @$CONFIG_FILE
17fi
18
19# Wait for connector to start
20sleep 5
21
22# Check status
23STATUS=$(curl -s "$CONNECT_HOST/connectors/$CONNECTOR_NAME/status" | jq -r '.connector.state')
24echo "Connector status: $STATUS"

Best Practices

For Production

• Use a load balancer for REST API requests across workers
• Implement retry logic for transient failures
• Monitor connector status continuously
• Use configuration management for connector configs
• Version control your connector configurations

For Security

• Enable authentication on REST API endpoints
• Use HTTPS for encrypted communication
• Store credentials in secure configuration providers
• Implement access controls for connector management

Mastering the REST API is essential for operating Kafka Connect. With POST, GET, PUT, and DELETE operations, you have complete control over your data pipelines. Combined with automation and monitoring, the REST API enables production-grade connector management.

This concludes Chapter 3 on installing and running Kafka Connect. You now have the knowledge to deploy Connect in both standalone and distributed modes, install connector plugins, and manage connectors through the REST API.