DEV Community

Scott Wang
Scott Wang

Posted on

MySQL 8 Kafka Connect Tutorial on Docker

#kafka #mysql #kafkaconnect #docker

In this tutorial, we will use docker-compose, MySQL 8 as examples to demonstrate Kafka Connector by using MySQL as the data source.

This tutorial is mainly based on the tutorial written on Kafka Connect Tutorial on Docker. However, the original tutorial is out-dated that it just won’t work if you followed it step by step. This is a refreshment of that tutorial, also to simplify things, we will get rid of the Avro set up as it serves no purpose to demonstrate Kafka Connector. Enough said, let’s begins.

Preparation

First, you will need to download MYSQL Connector Driver, this can be found
MySQL Connector Driver

You will also need to download the JDBC plugins at
Confluent JDBC plugins

Unzip both mysql-connector-java-8.0.22.tar.gz and confluentinc-kafka-connect-jdbc-10.0–2.1.zip. Create a jars directory, move mysql-connector-java-8.0.22.jar and all the .jar files in onfluentinc-kafka-connect-jdbc-10.0–2.1/lib/ directory to the jars directory.

docker-compose file

Here is the docker-compose file that contains everything you need to run this tutorial

version: '2'
services:
  mysql:
    privileged: true
    ports:
      - 3306:3306
    environment:
      MYSQL_ROOT_PASSWORD: test
    image: mysql:8.0

  zookeeper:
    image: confluentinc/cp-zookeeper:5.0.0
    privileged: true
    ports:
      - 32181:32181
    environment:
      ZOOKEEPER_CLIENT_PORT: 32181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:5.0.0
    ports:
      - 29092:29092
    links:
      - zookeeper
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:32181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  kafka-connector-mysql:
    image: confluentinc/cp-kafka-connect:latest
    ports:
      - 28083:28083
    links:
      - kafka
      - zookeeper
      - mysql
    environment:
      CONNECT_BOOTSTRAP_SERVERS: kafka:29092
      CONNECT_REST_PORT: 28083
      CONNECT_GROUP_ID: "quickstart-avro"
      CONNECT_CONFIG_STORAGE_TOPIC: "quickstart-avro-config"
      CONNECT_OFFSET_STORAGE_TOPIC: "quickstart-avro-offsets"
      CONNECT_STATUS_STORAGE_TOPIC: "quickstart-avro-status"
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_REST_ADVERTISED_HOST_NAME: "localhost"
      CONNECT_LOG4J_ROOT_LOGLEVEL: DEBUG
      CONNECT_PLUGIN_PATH: "/usr/share/java,/etc/kafka-connect/jars"
    volumes:
      - $PWD/jars:/etc/kafka-connect/jars
Enter fullscreen mode Exit fullscreen mode

Before you run the docker-compose, make sure your file structure looks like
Alt Text

Let the fun begins

Let the fun begins with starting mysql service

docker-compose up -d mysql
docker exec -it kafka-connector_mysql_1 bash
Enter fullscreen mode Exit fullscreen mode

Execute the following queries by using MySQL cli mysql -uroot -ptest

CREATE DATABASE IF NOT EXISTS connect_test;
USE connect_test;

CREATE TABLE IF NOT EXISTS test (
  id serial NOT NULL PRIMARY KEY,
  name varchar(100),
  email varchar(200),
  department varchar(200),
  modified timestamp default CURRENT_TIMESTAMP NOT NULL,
  INDEX `modified_index` (`modified`)
);

INSERT INTO test (name, email, department) VALUES ('alice', 'alice@abc.com', 'engineering');
INSERT INTO test (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
INSERT INTO test (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
INSERT INTO test (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
INSERT INTO test (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
INSERT INTO test (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
INSERT INTO test (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
INSERT INTO test (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
INSERT INTO test (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
INSERT INTO test (name, email, department) VALUES ('bob', 'bob@abc.com', 'sales');
exit;
Enter fullscreen mode Exit fullscreen mode

Let’s start zookeeper and kafka

docker-compose up -d zookeeper kafka
Enter fullscreen mode Exit fullscreen mode

wait for few seconds, to make sure that all services are up and running.

docker ps
Enter fullscreen mode Exit fullscreen mode

Once the zookeeper, Kafka and mysql are all up and running, let’s prepare our final course, confluent kafka-connect
First, let’s create the topics that will be used by the connector

docker-compose run --rm kafka kafka-topics --create --topic quickstart-avro-offsets --partitions 1 --replication-factor 1 --if-not-exists --zookeeper zookeeper:32181 --config cleanup.policy=compact

docker-compose run --rm kafka kafka-topics --create --topic quickstart-avro-config --partitions 1 --replication-factor 1 --if-not-exists --zookeeper zookeeper:32181 --config cleanup.policy=compact

docker-compose run --rm kafka kafka-topics --create --topic quickstart-avro-status --partitions 1 --replication-factor 1 --if-not-exists --zookeeper zookeeper:32181 --config cleanup.policy=compact
Enter fullscreen mode Exit fullscreen mode

Finally, let’s start the kafka-connect

docker-compose up -d kafka-connector-mysql
Enter fullscreen mode Exit fullscreen mode

Give it few seconds, and let’s create the JDBC Source connector by making a REST API call to the kafka connector service.

curl -X POST \
  -H "Content-Type: application/json" \
  --data '{ "name": "quickstart-jdbc-source", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": 1, "connection.url": "jdbc:mysql://mysql:3306/connect_test", "connection.user": "root", "connection.password": "test", "mode": "incrementing", "incrementing.column.name": "id", "timestamp.column.name": "modified", "topic.prefix": "quickstart-jdbc-", "poll.interval.ms": 1000 } }' \
  http://localhost:28083/connectors
Enter fullscreen mode Exit fullscreen mode

wait for few seconds, and check the status to make sure it is RUNNING

curl -s -X GET http://localhost:28083/connectors/quickstart-jdbc-source/status
Enter fullscreen mode Exit fullscreen mode

You should see something similar to

{"name":"quickstart-jdbc-source","connector":{"state":"RUNNING","worker_id":"localhost:28083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"localhost:28083"}],"type":"source"}%
Enter fullscreen mode Exit fullscreen mode

That’s yet, let’s verify our work by running a Kafka consumer,

docker-compose run --rm kafka kafka-console-consumer --bootstrap-server kafka:29092  --topic quickstart-jdbc-test --from-beginning
Enter fullscreen mode Exit fullscreen mode

and you should see the following logs show up

{"schema":{"type":"struct","fields":[{"type":"int64","optional":false,"field":"id"},{"type":"string","optional":true,"field":"name"},{"type":"string","optional":true,"field":"email"},{"type":"string","optional":true,"field":"department"},{"type":"int64","optional":false,"name":"org.apache.kafka.connect.data.Timestamp","version":1,"field":"modified"}],"optional":false,"name":"test"},"payload":{"id":18,"name":"bob","email":"bob@abc.com","department":"sales","modified":1606418867000}}
{"schema":{"type":"struct","fields":[{"type":"int64","optional":false,"field":"id"},{"type":"string","optional":true,"field":"name"},{"type":"string","optional":true,"field":"email"},{"type":"string","optional":true,"field":"department"},{"type":"int64","optional":false,"name":"org.apache.kafka.connect.data.Timestamp","version":1,"field":"modified"}],"optional":false,"name":"test"},"payload":{"id":19,"name":"bob","email":"bob@abc.com","department":"sales","modified":1606418867000}}
{"schema":{"type":"struct","fields":[{"type":"int64","optional":false,"field":"id"},{"type":"string","optional":true,"field":"name"},{"type":"string","optional":true,"field":"email"},{"type":"string","optional":true,"field":"department"},{"type":"int64","optional":false,"name":"org.apache.kafka.connect.data.Timestamp","version":1,"field":"modified"}],"optional":false,"name":"test"},"payload":{"id":20,"name":"bob","email":"bob@abc.com","department":"sales","modified":1606418867000}}
^CProcessed a total of 10 messages
Enter fullscreen mode Exit fullscreen mode

That’s yet. To tear it all down, simply run

docker-compose down
Enter fullscreen mode Exit fullscreen mode

Enjoy.

Top comments (2)

Collapse
 
zakrian07 profile image
zakrian07 •

docker ps always show starting and curl request not success ?
any idea how to fix it?

Collapse
 
nhatdiec profile image
nhatdiec • • Edited

Hi Scott Wang,
I'm newbie on kafka, I followed this article and have error as image below. Am I missing any files?
dev-to-uploads.s3.amazonaws.com/up...

Read next

drupaladmin profile image

Migrating to ClickHouse: Releem's Journey

Roman Agabekov -

brayanmonteiroo profile image

Como criar um ambiente Docker com PHP, Apache, PostgreSQL e pgAdmin

Brayan Monteiro -

amatisse profile image

Docker Demystified: A Beginner's Guide to Containerization 🐳

AMatisse -

rajarajacholank profile image

docker compose : top level object must be mapping

Rajarajacholan -