Chapter 8. Using Streams for Apache Kafka with Kafka Connect

Procedure

1. Edit the /opt/kafka/config/connect-standalone.properties Kafka Connect configuration file and set bootstrap.server to point to your Kafka brokers. For example:

   bootstrap.servers=kafka0.my-domain.com:9092,kafka1.my-domain.com:9092,kafka2.my-domain.com:9092

2. Start Kafka Connect with the configuration file and specify one or more connector configurations.

   su - kafka
   /opt/kafka/bin/connect-standalone.sh /opt/kafka/config/connect-standalone.properties connector1.properties
   [connector2.properties ...]

3. Verify that Kafka Connect is running.

   jcmd | grep ConnectStandalone

Running the cluster

1. Edit the /opt/kafka/config/connect-distributed.properties Kafka Connect configuration file on all Kafka Connect worker nodes.

   • Set the bootstrap.server option to point to your Kafka brokers.
   • Set the group.id option.
   • Set the config.storage.topic option.
   • Set the offset.storage.topic option.

   • Set the status.storage.topic option.

     For example:

     bootstrap.servers=kafka0.my-domain.com:9092,kafka1.my-domain.com:9092,kafka2.my-domain.com:9092
     group.id=my-group-id
     config.storage.topic=my-group-id-configs
     offset.storage.topic=my-group-id-offsets
     status.storage.topic=my-group-id-status

2. Start the Kafka Connect workers with the /opt/kafka/config/connect-distributed.properties configuration file on all Kafka Connect nodes.

   su - kafka
   /opt/kafka/bin/connect-distributed.sh /opt/kafka/config/connect-distributed.properties

3. Verify that Kafka Connect is running.

   jcmd | grep ConnectDistributed

4. Use the Kafka Connect REST API to manage connectors.

Procedure

1. Prepare a JSON payload with the connector configuration. For example:

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

2. Send a POST request to <KafkaConnectAddress>:8083/connectors to create the connector. The following example uses curl:

   curl -X POST -H "Content-Type: application/json" --data @sink-connector.json http://connect0.my-domain.com:8083/connectors

3. Verify that the connector was deployed by sending a GET request to <KafkaConnectAddress>:8083/connectors. The following example uses curl:

   curl http://connect0.my-domain.com:8083/connectors

Deleting connectors

1. Verify that the connector exists by sending a GET request to <KafkaConnectAddress>:8083/connectors/<ConnectorName>. The following example uses curl:

   curl http://connect0.my-domain.com:8083/connectors

2. To delete the connector, send a DELETE request to <KafkaConnectAddress>:8083/connectors. The following example uses curl:

   curl -X DELETE http://connect0.my-domain.com:8083/connectors/my-connector

3. Verify that the connector was deleted by sending a GET request to <KafkaConnectAddress>:8083/connectors. The following example uses curl:

   curl http://connect0.my-domain.com:8083/connectors