Using Kafka Connect

1. Understanding Connect Framework

Kafka Connect is a tool for scalably and reliably streaming data between Apache Kafka and other systems. It provides a framework to connect Kafka with external systems like databases, file systems, and cloud services.

Component	Description	Role
Source Connector	Reads data from external systems and writes to Kafka topics	Producer of data into Kafka
Sink Connector	Reads data from Kafka topics and writes to external systems	Consumer of data from Kafka
Worker	Runs the connectors and tasks	Runtime environment for Connectors
Connector	Defines the configuration and behavior of a data flow	Configuration and logic for data movement
Task	Represents a unit of work within a connector	Actual execution unit for data transfer

Example: Source Connector Configuration

{
  "name": "mysql-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": 1,
    "connection.url": "jdbc:mysql://localhost:3306/mydb",
    "connection.user": "user",
    "connection.password": "password",
    "table.whitelist": "users",
    "topic.prefix": "mysql-"
  }
}

Example: Sink Connector Configuration

{
  "name": "elasticsearch-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": 1,
    "topics": "mysql-users",
    "connection.url": "http://localhost:9200",
    "type.name": "kafka-connect"
  }
}

Key Features of Kafka Connect

Highly scalable and fault-tolerant
Supports distributed and standalone modes
Provides REST API for managing connectors
Handles data transformation and schema evolution
Integrates with various external systems via connectors

2. Running Standalone Mode

Kafka Connect can be run in standalone mode, which uses a single worker process to manage all connectors and tasks.

Configuration Property	Default Value	Description
bootstrap.servers	`localhost:9092`	Comma-separated list of Kafka broker addresses
key.converter	`org.apache.kafka.connect.json.JsonConverter`	Converter used for keys in Kafka messages
value.converter	`org.apache.kafka.connect.json.JsonConverter`	Converter used for values in Kafka messages
offset.storage.topic	`connect-offsets`	Topic to store connector offsets
config.storage.topic	`connect-configs`	Topic to store connector configurations
status.storage.topic	`connect-status`	Topic to store connector status

Example: Standalone Configuration File

bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
offset.storage.topic=connect-offsets
config.storage.topic=connect-configs
status.storage.topic=connect-status

Note: Standalone mode is suitable for development and testing but not recommended for production environments due to lack of fault tolerance.

3. Running Distributed Mode

Property	Default Value	Description
group.id	`kafka-connect`	Unique identifier for the Connect worker group
config.storage.topic	`connect-configs`	Topic to store connector configurations
offset.storage.topic	`connect-offsets`	Topic to store connector offsets
status.storage.topic	`connect-status`	Topic to store connector status
key.converter	`org.apache.kafka.connect.json.JsonConverter`	Converter for key data types
value.converter	`org.apache.kafka.connect.json.JsonConverter`	Converter for value data types

Example: Distributed Mode Configuration

group.id=connect-cluster
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
rest.port=8083

Note: In distributed mode, Connect workers coordinate to manage connector tasks. Each worker must be part of the same group to share configuration and status information.

Steps to Start Distributed Mode

Start ZooKeeper (if not already running)
Start Kafka brokers
Configure Connect worker properties (e.g., connect-distributed.properties)
Start the Connect worker process with kafka-run-class.sh
Verify that the Connect REST API is accessible at http://localhost:8083

Key Features of Distributed Mode

High availability through worker coordination
Automatic task assignment and rebalancing
Centralized configuration management
REST API for connector lifecycle management

4. Creating Source Connectors

Property	Description	Example Value
name	Name of the connector	`source-connector-1`
config.topic	Topic to publish data to	`source-topic`
config.connector.class	Class name of the connector	`org.apache.kafka.connect.file.FileStreamSourceConnector`
config.tasks.max	Maximum number of tasks to run	`1`
config.file	Path to the file to read from	`/tmp/test.txt`

Example: Creating a Source Connector

{
  "name": "source-connector-1",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": "1",
    "file": "/tmp/test.txt",
    "topic": "source-topic"
  }
}

Note: Ensure the file path specified in config.file exists and is readable by the Kafka Connect worker.

5. Creating Sink Connectors

Sink connectors in Kafka Connect are responsible for moving data from Kafka topics into external systems like databases, search engines, or file systems.

Property	Values	Description
connector.class	`org.apache.kafka.connect.file.FileStreamSinkConnector`	The Java class name of the connector to use.
tasks.max	`1`	The maximum number of tasks to create for this connector.
topics	`my-topic`	A comma-separated list of Kafka topics to consume from.
file	`/tmp/my-file.txt`	The file path where data will be written.

Example: Creating a Sink Connector

{
  "name": "file-sink-connector",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "tasks.max": "1",
    "topics": "my-topic",
    "file": "/tmp/my-file.txt"
  }
}

Note: Ensure that the Kafka Connect worker has appropriate permissions to read from Kafka and write to the specified file or external system.

6. Configuring Connector Properties

Property	Description	Default Value	Notes
connector.class	Class name of the connector to use	None	Required field for all connectors
tasks.max	Maximum number of tasks to run for this connector	1	Can be increased to scale processing
name	Name of the connector instance	None	Must be unique within the Kafka Connect cluster
transforms	Transforms to apply to records	None	Used for data transformation before sending to Kafka
errors.log.enable	Whether to log errors to the error topic	false	Requires `errors.topic.name` to be set
errors.log.include.messages	Whether to include the full record in error logs	false	Useful for debugging but can increase log size

Example: Basic Connector Configuration

{
  "name": "my-source-connector",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": 2,
    "file": "/tmp/test.txt",
    "topic": "my-topic"
  }
}

Example: Connector with Transformations

{
  "name": "my-connector-with-transforms",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": 1,
    "file": "/tmp/test.txt",
    "topic": "my-topic",
    "transforms": "maskFields",
    "transforms.maskFields.type": "org.apache.kafka.connect.transforms.MaskField$Value",
    "transforms.maskFields.field": "password"
  }
}

Note: Connector properties can be set either in a configuration file or via the Kafka Connect REST API. Configuration values are case-sensitive and must match exactly.

7. Managing Connectors via REST API

Example: List all connectors

curl -X GET http://localhost:8083/connectors

Endpoint	Method	Description
/connectors	GET	List all connectors
/connectors/{name}	GET	Get connector configuration
/connectors/{name}/config	PUT	Update connector configuration
/connectors/{name}/pause	PUT	Pause a running connector
/connectors/{name}/resume	PUT	Resume a paused connector
/connectors/{name}/delete	DELETE	Delete a connector

Example: Create a new connector

{
  "name": "my-source-connector",
  "config": {
    "connector.class": "FileStreamSourceConnector",
    "tasks.max": "1",
    "file": "/tmp/test.txt",
    "topic": "source-topic"
  }
}

Example: Update connector config

curl -X PUT http://localhost:8083/connectors/my-source-connector/config \
  -H "Content-Type: application/json" \
  -d '{"tasks.max": "2"}'

Note: The REST API is available at the Kafka Connect worker's HTTP port, typically 8083.

Warning: Updating connector configuration while it's running may cause data loss or inconsistency. Always pause the connector before updating its config.

8. Listing Connectors

Property	Description	Default Value	Notes
connectors.list	Lists all currently running connectors	None	Available via REST API endpoint `/connectors`

Example: Listing Connectors via REST API

curl -X GET http://localhost:8083/connectors

Note: The response is a JSON array containing the names of all active connectors.

9. Checking Connector Status

Property	Description	Values
connector.status	Current status of the connector	`UNASSIGNED`, `RUNNING`, `PAUSED`, `FAILED`
tasks.status	Status of each task associated with the connector	`UNASSIGNED`, `RUNNING`, `PAUSED`, `FAILED`
connector.name	Name of the connector instance	String identifier

Example: Retrieve connector status via REST API

curl -X GET http://localhost:8083/connectors/my-connector/status

Example: Sample response from connector status endpoint

{
  "name": "my-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "192.168.1.100:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "192.168.1.100:8083"
    }
  ],
  "type": "source"
}

10. Pausing and Resuming Connectors

Connectors in Kafka Connect can be paused and resumed to control their execution state without stopping the entire Connect worker.

Operation	Description	API Endpoint	Notes
pause	Pauses a running connector and its tasks.	`PUT /connectors/{connectorName}/pause`	Connector tasks are stopped but configuration remains.
resume	Resumes a paused connector and restarts its tasks.	`PUT /connectors/{connectorName}/resume`	Tasks are restarted from the last committed offset.

Example: Pausing a Connector

curl -X PUT http://localhost:8083/connectors/my-source-connector/pause

Example: Resuming a Connector

curl -X PUT http://localhost:8083/connectors/my-source-connector/resume

Note: Pausing a connector does not stop the underlying tasks immediately. Tasks may continue running until their next commit cycle.

11. Deleting Connectors

Property	Description	Value
connector.name	Name of the connector to delete	`my-connector`

Example: Delete a Connector

curl -X DELETE http://localhost:8083/connectors/my-connector

Note: Deleting a connector stops all tasks and removes the connector configuration from Kafka Connect.