{"id":740,"date":"2020-04-21T17:37:14","date_gmt":"2020-04-21T16:37:14","guid":{"rendered":"http:\/\/www.igfasouza.com\/blog\/?p=740"},"modified":"2021-05-20T14:22:07","modified_gmt":"2021-05-20T13:22:07","slug":"kafka-connect","status":"publish","type":"post","link":"http:\/\/www.igfasouza.com\/blog\/kafka-connect\/","title":{"rendered":"Kafka Connect"},"content":{"rendered":"

\"\"<\/p>\n

How are you?<\/b><\/p>\n

This blog post is part of my series of posts regarding Kafka Connect.
\nIf you’re not familiar with Kafka, I suggest you have a look at some of my previous post;<\/p>\n

What is Kafka?<\/a>
\nKafka Connect Overview<\/a>
\nKafka Connector Architecture<\/a><\/p>\n

This post is a collection of links, videos, tutorials, blogs and books that I found mixed with my opinion. <\/p>\n

Table of contents<\/h2>\n

1. Source Standalone mode
\n2. Source Distributed mode
\n3. Sink
\n4. Don\u2019t use Docker composer
\n5. Lensesio
\n6. Strimzi
\n7. Debezium
\n8. JDBC
\n9. Link<\/p>\n

1. Source Standalone mode<\/h2>\n

Standalone mode is the best way to get started with minimum config and infrastructure setup. In this section, we will see how to configure a connector in Standalone mode and transfer file content to Kafka topic.<\/p>\n

Here I\u2019m going to show how to use the FileSource Connector<\/a><\/p>\n

Let\u2019s create a Docker compose file. \u201cdocker-compose.yml\u201d<\/p>\n

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
<\/div><\/td>
zookeeper:
\n    image: zookeeper:3.4.9
\n    restart: unless-stopped
\n    hostname: zookeeper
\n    container_name: zookeeper
\n    ports:
\n      - "2181:2181"<\/span>
\n    environment:
\n        ZOO_MY_ID: 1<\/span>
\n        ZOO_PORT: 2181<\/span>
\n        ZOO_SERVERS: server.1=<\/span>zookeeper:2888<\/span>:3888<\/span>
\n  kafka:
\n    image: confluentinc\/cp-kafka:5.1.0
\n    hostname: kafka
\n    container_name: kafka
\n    ports:
\n      - "9092:9092"<\/span>
\n    environment:
\n      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,<\/span>PLAINTEXT_HOST:PLAINTEXT
\n      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT:\/\/kafka:9092<\/span>,<\/span>PLAINTEXT_HOST:\/\/localhost:29092<\/span>
\n      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"<\/span>
\n      KAFKA_BROKER_ID: 1<\/span>
\n      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1<\/span>
\n    depends_on:
\n      - zookeeper
\n  kafka-connect:
\n    image: confluentinc\/cp-kafka-connect:5.1.0
\n    hostname: kafka-connect
\n    container_name: kafka-connect
\n    ports:
\n      - "8083:8083"<\/span>
\n    environment:
\n      CONNECT_BOOTSTRAP_SERVERS: "kafka:9092"<\/span>
\n      CONNECT_REST_ADVERTISED_HOST_NAME: connect
\n      CONNECT_REST_PORT: 8083<\/span>
\n      CONNECT_GROUP_ID: compose-connect-group
\n      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
\n      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
\n      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
\n      CONNECT_KEY_CONVERTER: org.apache<\/span>.kafka<\/span>.connect<\/span>.json<\/span>.JsonConverter<\/span>
\n      CONNECT_VALUE_CONVERTER: org.apache<\/span>.kafka<\/span>.connect<\/span>.json<\/span>.JsonConverter<\/span>
\n      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"<\/span>
\n      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"<\/span>
\n      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: "1"<\/span>
\n      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: "1"<\/span>
\n      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: "1"<\/span>
\n      CONNECT_PLUGIN_PATH: '\/usr\/share\/java,\/etc\/kafka-connect\/jars'<\/span>
\n      CONNECT_CONFLUENT_TOPIC_REPLICATION_FACTOR: 1<\/span>
\n    depends_on:
\n      - zookeeper
\n      - kafka<\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

In the same directory where we have created the yaml file execute the following command to start Kafka cluster. When the below command runs for the very first time it downloads the image. Once the image is downloaded it creates a Kafka cluster.<\/p>\n

1
<\/div><\/td>
docker-compose up<\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

For starting any Kafka connect cluster we require – workers config and connector (file-stream) config.
\nCreate two files: workers-config.properties and file-stream-connector-properties.<\/p>\n

Workers-config.properties:<\/strong><\/p>\n

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
<\/div><\/td>
bootstrap.servers<\/span>=<\/span>127.0.0.1:9092<\/span>
\nkey.converter<\/span>=<\/span>org.apache<\/span>.kafka<\/span>.connect<\/span>.json<\/span>.JsonConverter<\/span>
\nkey.converter<\/span>.schemas<\/span>.enable<\/span>=<\/span>false
\nvalue.converter<\/span>=<\/span>org.apache<\/span>.kafka<\/span>.connect<\/span>.json<\/span>.JsonConverter<\/span>
\nvalue.converter<\/span>.schemas<\/span>.enable<\/span>=<\/span>false
\n# we always leave the internal key to JsonConverter<\/span>
\ninternal.key<\/span>.converter<\/span>=<\/span>org.apache<\/span>.kafka<\/span>.connect<\/span>.json<\/span>.JsonConverter<\/span>
\ninternal.key<\/span>.converter<\/span>.schemas<\/span>.enable<\/span>=<\/span>false
\ninternal.value<\/span>.converter<\/span>=<\/span>org.apache<\/span>.kafka<\/span>.connect<\/span>.json<\/span>.JsonConverter<\/span>
\ninternal.value<\/span>.converter<\/span>.schemas<\/span>.enable<\/span>=<\/span>false
\n# Rest API<\/span>
\nrest.port<\/span>=<\/span>8086<\/span>
\nrest.host<\/span>.name<\/span>=<\/span>127.0.0.1
\n# this config is only for standalone workers<\/span>
\noffset.storage<\/span>.file<\/span>.filename<\/span>=<\/span>standalone.offsets<\/span>
\noffset.flush<\/span>.interval<\/span>.ms<\/span>=<\/span>10000<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

File-stream-connector-properties:<\/strong><\/p>\n

1
2
3
4
5
6
7
<\/div><\/td>
# These are standard kafka connect parameters, need for ALL connectors<\/span>
\nname=<\/span>file-stream-kafka-connect-standalone
\nconnector.class<\/span>=<\/span>org.apache<\/span>.kafka<\/span>.connect<\/span>.file<\/span>.FileStreamSourceConnector<\/span>
\ntasks.max<\/span>=<\/span>1<\/span>
\nfile<\/span>\/FileStreamSourceConnector.java<\/span>
\nfile<\/span>=<\/span>source-input<\/span>.txt<\/span>
\ntopic=<\/span>kafka-connect-standalone<\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

You should see three files in the folder; docker-compose.yml, file-stream-connector-properties and workers-config.properties<\/p>\n

Create an input file source-input.txt, the content of the file is transferred to Kafka topic.<\/p>\n

1
<\/div><\/td>
touch source-input<\/span>.txt<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

Mount a host directory in a docker container: Make sure we are in the directory where we have created the files. After mount, we automatically switch to cp-kafka-connect.<\/p>\n

1
<\/div><\/td>
docker run --rm -it -v "$(pwd)"<\/span>:\/kafka-connect\/ --net=<\/span>host confluentinc\/cp-kafka-connect<\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

This docker image expect the connector in this folder;
\n\/etc\/kafka-connect\/jars<\/p>\n

Create Kafka topic “kafka-connect-standalone” with 3 partitions and replication factor 1.<\/p>\n

1
<\/div><\/td>
kafka-topics --create --topic kafka-connect-standalone --partitions 3<\/span> --replication-factor 1<\/span> --zookeeper 127.0.0.1:2181<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

Create standalone connector using workers-config.properties and file-stream-connector-properties.<\/p>\n

1
<\/div><\/td>
connect-standalone workers-config.properties<\/span> file-stream-connector-properties<\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

Now you can test: Open file source-input.txt and type some message to it & save it. The message should have been transferred to Kafka topic. <\/p>\n

What happened in the background: We wrote data to the source file and Kafka connect standalone pushed the data to topic. No programming, just configs.<\/p>\n

Now stop Kafka Connect (press Ctrl\/Command + C). Once Kafka connect gracefully stopped, list all files in the given directory. We have a new guest in the form of a file “standalone.offsets”.
\nThis file is created by Kafka connect to keep track of from where it should resume reading messages from the source file on re-starts.
\nWhen Kafka connect starts again it should resume reading without publishing duplicate messages to the topic.<\/p>\n

Try to execute the above command (connect-standalone workers-config.properties file-stream-connector-properties) again and validate if you do not have a duplicate message.<\/p>\n

2. Source Distributed mode<\/h2>\n

Now let\u2019s configure a connector in distributed mode.<\/p>\n

Create Kafka topic “kafka-connect-distibuted” with 3 partitions and replication factor 1.<\/p>\n

1
<\/div><\/td>
kafka-topics --create --topic kafka-connect-distributed --partitions 3<\/span> --replication-factor 1<\/span> --zookeeper 127.0.0.1:2181<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

Create a config file \u201cconnect-source.json\u201d.<\/p>\n

1
2
3
4
5
6
7
8
9
<\/div><\/td>
name=<\/span>file-stream-kafka-connect-distributed
\nconnector.class<\/span>=<\/span>org.apache<\/span>.kafka<\/span>.connect<\/span>.file<\/span>.FileStreamSourceConnector<\/span>
\ntasks.max<\/span>=<\/span>1<\/span>
\nfile<\/span>=<\/span>source-input<\/span>.txt<\/span>
\ntopic=<\/span>kafka-connect-distributed
\nkey.converter<\/span>=<\/span>org.apache<\/span>.kafka<\/span>.connect<\/span>.json<\/span>.JsonConverter<\/span>
\nkey.converter<\/span>.schemas<\/span>.enable<\/span>=<\/span>true
\nvalue.converter<\/span>=<\/span>org.apache<\/span>.kafka<\/span>.connect<\/span>.json<\/span>.JsonConverter<\/span>
\nvalue.converter<\/span>.schemas<\/span>.enable<\/span>=<\/span>true<\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

and run;<\/p>\n

1
<\/div><\/td>
curl -d @<<\/span>path-to-config-file><\/span>\/connect-source.json<\/span> -H "Content-Type: application\/json"<\/span> -X POST http:\/\/localhost:8083<\/span>\/connectors<\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

Now write some message in the source file and once the file is saved, all messages are posted in topic “kafka-connect-distributed”. <\/p>\n

check for the message copied from file to topic by FileSourceConnector. Note that messages are stored in JSON format as the connector topic created earlier with config value.converter.schemas.enable=true.<\/p>\n

We can run the command to check the topic;<\/p>\n

1
<\/div><\/td>
kafka-console-consumer --topic kafka-connect-distributed --from-beginning --bootstrap-server 127.0.0.1:9092<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

3. Sink<\/h2>\n

Now we need a sink example; Let\u2019s look at a JDBC one. MondoBD.
\nWe can add this to the docker composer file.<\/p>\n

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
<\/div><\/td>
 mongo-db:
\n    image: mongo:4.0.5
\n    hostname: mongo-db
\n    container_name: mongo-db
\n    expose:
\n      - "27017"<\/span>
\n    ports:
\n      - "27017:27017"<\/span>
\n    command: --bind_ip_all --smallfiles
\n    volumes:
\n      - .\/mongo-db:\/data
\n  mongoclient:
\n    image: mongoclient\/mongoclient:2.2.0
\n    container_name: mongoclient
\n    hostname: mongoclient
\n    depends_on:
\n      - mongo-db
\n    ports:
\n      - 3000<\/span>:3000<\/span>
\n    environment:
\n      MONGO_URL: "mongodb:\/\/mongo-db:27017"<\/span>
\n      PORT: 3000<\/span>
\n    expose:
\n      - "3000"<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

and run;<\/p>\n

1
<\/div><\/td>
curl -d @<<\/span>path-to-config file><\/span>\/connect-mongodb-sink.json<\/span> -H "Content-Type: application\/json"<\/span> -X POST http:\/\/localhost:8083<\/span>\/connectors<\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

Create a connect-mongodb-sink.json<\/p>\n

1
2
3
4
5
6
7
8
9
10
11
12
13
14
<\/div><\/td>
{<\/span>
\n    "name"<\/span>: "mongodb-sink"<\/span>,<\/span>
\n    "config"<\/span>: {<\/span>
\n        "connector.class"<\/span>: "at.grahsl.kafka.connect.mongodb.MongoDbSinkConnector"<\/span>,<\/span>
\n        "tasks.max"<\/span>: 1<\/span>,<\/span>
\n        "topics"<\/span>: "connect-custom"<\/span>,<\/span>
\n        "mongodb.connection.uri"<\/span>: "mongodb:\/\/mongo-db\/test?retryWrites=true"<\/span>,<\/span>
\n        "mongodb.collection"<\/span>: "MyCollection"<\/span>,<\/span>
\n        "key.converter"<\/span>: "org.apache.kafka.connect.json.JsonConverter"<\/span>,<\/span>
\n        "key.converter.schemas.enable"<\/span>: false,<\/span>
\n        "value.converter"<\/span>: "org.apache.kafka.connect.json.JsonConverter"<\/span>,<\/span>
\n        "value.converter.schemas.enable"<\/span>: false
\n    }<\/span>
\n}<\/span><\/div><\/td><\/tr><\/tbody><\/table><\/div>\n

We have the following MongoDB-specific properties here:<\/p>\n