Kafka Connect Quickstart

This guide will help with end-to-end local setup for Kafka Connect.

Sink Connector which will fetch records from a Kafka topic (kctopic_for_sink) and output its value on console.

• Assumptions: Java8, Maven, basic knowledge of Kafka
• Kafka version used for this guide: kafka_2.10–0.10.2.1
• Feel free to upgrade required libs to latest versions based on specific project requirements

1. Download Kafka from apache kafka

Create these 2 extra files

• connect-distributed-local.properties under “kafka_2.10–0.10.2.1/config/”

bootstrap.servers=localhost:9092
group.id=local-connect-cluster

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

offset.storage.topic=connect-local-stg-offsets
config.storage.topic=connect-local-stg-configs
status.storage.topic=connect-local-stg-status

consumer.max.poll.records=1
consumer.enable.auto.commit=false
consumer.auto.offset.reset=latest

• create_local_topics.sh under “kafka_2.10–0.10.2.1/” (used to create topics required to run Kafka-connect, you can also create these topics one by one and skip this .sh file)

#!/bin/sh
./bin/kafka-topics.sh — list — zookeeper localhost:2181
./bin/kafka-topics.sh — create — zookeeper localhost:2181 — replication-factor 1 — partitions 1 — topic connect-local-stg-offsets
./bin/kafka-topics.sh — create — zookeeper localhost:2181 — replication-factor 1 — partitions 1 — topic connect-local-stg-configs
./bin/kafka-topics.sh — create — zookeeper localhost:2181 — replication-factor 1 — partitions 1 — topic connect-local-stg-status
./bin/kafka-topics.sh — create — zookeeper localhost:2181 — replication-factor 1 — partitions 1 — topic kctopic_for_sink
./bin/kafka-topics.sh — list — zookeeper localhost:2181

These can be created separately, in case you are downloading Kafka from some other repo.

2. Build KCQuickstartSinkConnector jar from source code given below

Files of interest:

• KCQuickstartSinkConnector.java
• KCQuickstartSinkConnectorConfig.java
• KCQuickstartSinkTask.java

source code

3. Get console setup as shown in (to get a better view of commands running per screen)

4. Copy paste KCQuickstartSinkConnector jar to folder “kafka_2.10–0.10.2.1/libs/”

5. Start Kafka zookeeper

6. Start Kafka server

7. Create Kafka topics

8. Start Kafka Conosle Producer

9. Start Kafka Connect

10. Start a connector using REST API call curl -X POST -H "Content-Type: application/json" -d '{"name":"kc-local-quickstart","config":{"topics":"kctopic_for_sink","connector.class":"com.sample.kcquickstart.KCQuickstartSinkConnector","tasks.max":"1","key.converter":"org.apache.kafka.connect.storage.StringConverter","value.converter":"org.apache.kafka.connect.storage.StringConverter","key.converter.schemas.enable":"false","value.converter.schemas.enable":"false"}}' localhost:8083/connectors

11. Send few records to Kafka topic “kctopic_for_sink”, using Kafka-console producer

12. As there are new records on topic, connector should fetch these records and send it to SinkTask for processing

13. Check Kafka-Connect’s output console, records should show up here..

14. Now you have an end-to-end local setup for complete Kafka connect dev and test purpose.

Hope you enjoyed quick-start guide and learned something new. If yes, remember to hit the clap button :-)