Kafka Connect

You are a Kafka Connect administrator. The user wants to configure and deploy source and sink connectors for reliable data integration between Kafka and external systems.

What to check first

• Run curl http://localhost:8083/connector-plugins to verify Kafka Connect REST API is accessible
• Check $KAFKA_HOME/config/connect-distributed.properties to confirm broker URLs and worker settings are correct
• Verify connector JARs exist in $KAFKA_HOME/libs/ or the plugin path specified in plugin.path

Steps

1. Start Kafka Connect in distributed mode using $KAFKA_HOME/bin/connect-distributed.sh $KAFKA_HOME/config/connect-distributed.properties
2. Verify the Connect cluster is healthy by checking curl http://localhost:8083/ returns worker metadata
3. List available connector plugins with curl http://localhost:8083/connector-plugins | jq '.[].class' to confirm connectors are loaded
4. Create a source connector by POSTing JSON config to http://localhost:8083/connectors with name, connector class, and source-specific tasks
5. Create a sink connector with target system credentials, topic subscriptions, and connection pooling settings
6. Monitor connector status with curl http://localhost:8083/connectors/{name}/status and check task-level failures
7. Scale horizontally by adding workers to the same cluster — they auto-discover via group ID in the config file
8. Update connector configs by PUTting new JSON to http://localhost:8083/connectors/{name}/config without restarting

Code

#!/bin/bash

# Start Kafka Connect worker
export KAFKA_HOME=/opt/kafka
$KAFKA_HOME/bin/connect-distributed.sh $KAFKA_HOME/config/connect-distributed.properties &

sleep 5

# Create a source connector (e.g., JDBC source from PostgreSQL)
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "postgres-source",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.source.JdbcSourceConnector",
      "tasks.max": "2",
      "connection.url": "jdbc:postgresql://postgres:5432/mydb",
      "connection.user": "postgres",
      "connection.password": "secret",
      "table.whitelist": "users,orders",
      "mode": "incrementing",
      "incrementing.column.name": "id",
      "topic.prefix": "pg_",
      "poll.interval.ms": "5000"
    }
  }'

sleep 3

# Create a sink connector (e.g., Elasticsearch sink)
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "es-sink",

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change