Tutorial

Introduction to Debezium

Debezium is a distributed platform that converts information from your existing databases into event streams, enabling applications to detect, and immediately respond to row-level changes in the databases.

Debezium is built on top of Apache Kafka and provides a set of Kafka Connect compatible connectors. Each of the connectors works with a specific database management system (DBMS). Connectors record the history of data changes in the DBMS by detecting changes as they occur, and streaming a record of each change event to a Kafka topic. Consuming applications can then read the resulting event records from the Kafka topic.

By taking advantage of Kafka’s reliable streaming platform, Debezium makes it possible for applications to consume changes that occur in a database correctly and completely. Even if your application stops unexpectedly, or loses its connection, it does not miss events that occur during the outage. After the application restarts, it resumes reading from the topic from the point where it left off.

The tutorial that follows shows you how to deploy and use the Debezium MySQL connector with a simple configuration. For more information about deploying and using Debezium connectors, see the connector documentation.

Additional resources

Starting the services

Using Debezium requires three separate services: ZooKeeper, Kafka, and the Debezium connector service. In this tutorial, you will set up a single instance of each service using Docker and the Debezium container images.

To start the services needed for this tutorial, you must:

Considerations for running Debezium with Docker

This tutorial uses Docker and the Debezium container images to run the ZooKeeper, Kafka, Debezium, and MySQL services. Running each service in a separate container simplifies the setup so that you can see Debezium in action.
In a production environment, you would run multiple instances of each service to provide performance, reliability, replication, and fault tolerance. Typically, you would either deploy these services on a platform like OpenShift or Kubernetes that manages multiple Docker containers running on multiple hosts and machines, or you would install on dedicated hardware.

You should be aware of the following considerations for running Debezium with Docker:

  • The containers for ZooKeeper and Kafka are ephemeral.

ZooKeeper and Kafka would typically store their data locally inside the containers, which would require you to mount directories on the host machine as volumes. That way, when the containers are stopped, the persisted data remains. However, this tutorial skips this setup - when a container is stopped, all persisted data is lost. This way, cleanup is simple when you complete the tutorial.
For more information about storing persistent data, see the documentation for the container images.
- This tutorial requires you to run each service in a different container.

To avoid confusion, you will run each container in the foreground in a separate terminal. This way, all of the output of a container will be displayed in the terminal used to run it.
Docker also allows you to run a container in detached mode (with the -d option), where the container is started and the docker command returns immediately. However, detached mode containers do not display their output in the terminal. To see the output, you would need to use the docker logs —follow —name <container-name> command. For more information, see the Docker documentation.

Starting Zookeeper

ZooKeeper is the first service you must start.

Procedure

  1. Open a terminal and use it to start ZooKeeper in a container.

This command runs a new container using version 1.9 of the quay.io/debezium/zookeeper image:

1. $ docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 quay.io/debezium/zookeeper:1.9

-it

The container is interactive, which means the terminal’s standard input and output are attached to the container.

--rm

The container will be removed when it is stopped.

--name zookeeper

The name of the container.

-p 2181:2181 -p 2888:2888 -p 3888:3888

Maps three of the container’s ports to the same ports on the Docker host. This enables other containers (and applications outside of the container) to communicate with ZooKeeper.
If you use Podman, run the following command:s 1. $ sudo podman pod create —name=dbz —publish “9092,3306,8083” 2. $ sudo podman run -it —rm —name zookeeper —pod dbz quay.io/debezium/zookeeper:1.9
  1. Verify that ZooKeeper started and is listening on port 2181.

You should see output similar to the following:

1. Starting up in standalone mode 2. ZooKeeper JMX enabled by default 3. Using config: /zookeeper/conf/zoo.cfg 4. 2017-09-21 07:15:55,417 - INFO [main:QuorumPeerConfig@134] - Reading configuration from: /zookeeper/conf/zoo.cfg 5. 2017-09-21 07:15:55,419 - INFO [main:DatadirCleanupManager@78] - autopurge.snapRetainCount set to 3 6. 2017-09-21 07:15:55,419 - INFO [main:DatadirCleanupManager@79] - autopurge.purgeInterval set to 1 7. ... 8. port 0.0.0.0/0.0.0.0:2181 (1)
1 This line indicates that ZooKeeper is ready and listening on port 2181. The terminal will continue to show additional output as ZooKeeper generates it.

Starting Kafka

After starting ZooKeeper, you can start Kafka in a new container.
Debezium 1.9.6.Final has been tested against multiple versions of Kafka Connect. Please refer to the Debezium Test Matrix to determine compatibility between Debezium and Kafka Connect.

Procedure

  1. Open a new terminal and use it to start Kafka in a container.

This command runs a new container using version 1.9 of the quay.io/debezium/kafka image:

1. $ docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper quay.io/debezium/kafka:1.9

-it

The container is interactive, which means the terminal’s standard input and output are attached to the container.

--rm

The container will be removed when it is stopped.

--name kafka

The name of the container.

-p 9092:9092

Maps port 9092 in the container to the same port on the Docker host so that applications outside of the container can communicate with Kafka.

--link zookeeper:zookeeper

Tells the container that it can find ZooKeeper in the zookeeper container, which is running on the same Docker host.
If you use Podman, run the following command: 1. $ sudo podman run -it —rm —name kafka —pod dbz quay.io/debezium/kafka:1.9
In this tutorial, you will always connect to Kafka from within a Docker container. Any of these containers can communicate with the kafka container by linking to it. If you needed to connect to Kafka from outside of a Docker container, you would have to set the -e option to advertise the Kafka address through the Docker host (-e ADVERTISED_HOST_NAME= followed by either the IP address or resolvable host name of the Docker host).
2. Verify that Kafka started.

You should see output similar to the following:

1. ... 2. 2017-09-21 07:16:59,085 - INFO [main-EventThread:ZkClient@713] - zookeeper state changed (SyncConnected) 3. 2017-09-21 07:16:59,218 - INFO [main:Logging$class@70] - Cluster ID = LPtcBFxzRvOzDSXhc6AamA 4. ... 5. 2017-09-21 07:16:59,649 - INFO [main:Logging$class@70] - [Kafka Server 1], started (1)
1 The Kafka broker has successfully started and is ready for client connections. The terminal will continue to show additional output as Kafka generates it.

Starting a MySQL database

At this point, you have started ZooKeeper and Kafka, but you still need a database server from which Debezium can capture changes. In this procedure, you will start a MySQL server with an example database.

Procedure

  1. Open a new terminal, and use it to start a new container that runs a MySQL database server preconfigured with an inventory database.

This command runs a new container using version 1.9 of the quay.io/debezium/example-mysql image, which is based on the mysql:8.0 image. It also defines and populates a sample inventory database:

1. $ docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw quay.io/debezium/example-mysql:1.9

-it

The container is interactive, which means the terminal’s standard input and output are attached to the container.

--rm

The container will be removed when it is stopped.

--name mysql

The name of the container.

-p 3306:3306

Maps port 3306 (the default MySQL port) in the container to the same port on the Docker host so that applications outside of the container can connect to the database server.

-e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw

Creates a user and password that has the minimum privileges required by the Debezium MySQL connector.
If you use Podman, run the following command: 1. $ sudo podman run -it —rm —name mysql —pod dbz -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw quay.io/debezium/example-mysql:1.9
  1. Verify that the MySQL server starts.

The MySQL server starts and stops a few times as the configuration is modified. You should see output similar to the following:

1. ... 2. [System] [MY-010931] [Server] /usr/sbin/mysqld: ready for connections. Version: '8.0.27' socket: '/var/run/mysqld/mysqld.sock' port: 3306 MySQL Community Server - GPL. 3. [System] [MY-011323] [Server] X Plugin ready for connections. Bind-address: '::' port: 33060, socket: /var/run/mysqld/mysqlx.sock

Starting a MySQL command line client

After starting MySQL, you start a MySQL command line client so that you access the sample inventory database.

Procedure

  1. Open a new terminal, and use it to start the MySQL command line client in a container.

This command runs a new container using the mysql:8.0 image, and defines a shell command to run the MySQL command line client with the correct options:

1. $ docker run -it --rm --name mysqlterm --link mysql --rm mysql:8.0 sh -c 'exec mysql -h"$MYSQL_PORT_3306_TCP_ADDR" -P"$MYSQL_PORT_3306_TCP_PORT" -uroot -p"$MYSQL_ENV_MYSQL_ROOT_PASSWORD"'

-it

The container is interactive, which means the terminal’s standard input and output are attached to the container.

--rm

The container will be removed when it is stopped.

--name mysqlterm

The name of the container.

--link mysql

Links the container to the mysql container.
If you use Podman, run the following command: 1. $ sudo podman run -it —rm —name mysqlterm —pod dbz —rm mysql:5.7 sh -c ‘exec mysql -h 0.0.0.0 -uroot -pdebezium’
  1. Verify that the MySQL command line client started.

You should see output similar to the following:

``` 1. mysql: [Warning] Using a password on the command line interface can be insecure. 2. Welcome to the MySQL monitor. Commands end with ; or \g. 3. Your MySQL connection id is 9 4. Server version: 8.0.27 MySQL Community Server - GPL

  1. Copyright (c) 2000, 2021, Oracle and/or its affiliates.

  2. Oracle is a registered trademark of Oracle Corporation and/or its

  3. affiliates. Other names may be trademarks of their respective

  4. owners.

  5. Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.

  6. mysql> ```

  7. At the mysql> command prompt, switch to the inventory database:

1. mysql> use inventory; 3. List the tables in the database:

1. mysql> show tables; 2. +---------------------+ 3. | Tables_in_inventory | 4. +---------------------+ 5. | addresses | 6. | customers | 7. | geom | 8. | orders | 9. | products | 10. | products_on_hand | 11. +---------------------+ 12. 6 rows in set (0.00 sec) 4. Use the MySQL command line client to explore the database and view the pre-loaded data in the database.

For example:

1. mysql> SELECT * FROM customers; 2. +------+------------+-----------+-----------------------+ 3. | id | first_name | last_name | email | 4. +------+------------+-----------+-----------------------+ 5. | 1001 | Sally | Thomas | sally.thomas@acme.com | 6. | 1002 | George | Bailey | gbailey@foobar.com | 7. | 1003 | Edward | Walker | ed@walker.com | 8. | 1004 | Anne | Kretchmar | annek@noanswer.org | 9. +------+------------+-----------+-----------------------+ 10. 4 rows in set (0.00 sec)

Starting Kafka Connect

After starting MySQL and connecting to the inventory database with the MySQL command line client, you start the Kafka Connect service. This service exposes a REST API to manage the Debezium MySQL connector.

Procedure

  1. Open a new terminal, and use it to start the Kafka Connect service in a container.

This command runs a new container using the 1.9 version of the quay.io/debezium/connect image:

1. $ docker run -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link kafka:kafka --link mysql:mysql quay.io/debezium/connect:1.9

-it

The container is interactive, which means the terminal’s standard input and output are attached to the container.

--rm

The container will be removed when it is stopped.

--name connect

The name of the container.

-p 8083:8083

Maps port 8083 in the container to the same port on the Docker host. This enables applications outside of the container to use Kafka Connect’s REST API to set up and manage new container instances.

-e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses

Sets environment variables required by the Debezium image.

--link kafka:kafka --link mysql:mysql

Links the container to the containers that are running Kafka and the MySQL server.
If you use Podman, run the following command: 1. $ sudo podman run -it —rm —name connect —pod dbz -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses quay.io/debezium/connect:1.9
If you provide a —hostname command option then Kafka Connect REST API will not listen on the localhost interface. This can cause issues when the REST port is being exposed. If this is a problem then set environment variable REST_HOST_NAME=0.0.0.0 which will ensure that REST API will be accessible from all interfaces.
  1. Verify that Kafka Connect started and is ready to accept connections.

You should see output similar to the following:

1. ... 2. 2020-02-06 15:48:33,939 INFO || Kafka version: 3.0.0 [org.apache.kafka.common.utils.AppInfoParser] 3. ... 4. 2020-02-06 15:48:34,485 INFO || [Worker clientId=connect-1, groupId=1] Starting connectors and tasks using config offset -1 [org.apache.kafka.connect.runtime.distributed.DistributedHerder] 5. 2020-02-06 15:48:34,485 INFO || [Worker clientId=connect-1, groupId=1] Finished starting connectors and tasks [org.apache.kafka.connect.runtime.distributed.DistributedHerder] 2. Use the Kafka Connect REST API to check the status of the Kafka Connect service.

Kafka Connect exposes a REST API to manage Debezium connectors. To communicate with the Kafka Connect service, you can use the curl command to send API requests to port 8083 of the Docker host (which you mapped to port 8083 in the connect container when you started Kafka Connect).
These commands use localhost. If you are using a non-native Docker platform (such as Docker Toolbox), replace localhost with the IP address of your Docker host.

  1. Open a new terminal and check the status of the Kafka Connect service:

    1. $ curl -H "Accept:application/json" localhost:8083/ 2. {"version":"3.2.0","commit":"cb8625948210849f"} (1)
    1 The response shows that Kafka Connect version 3.2.0 is running.
    2. Check the list of connectors registered with Kafka Connect:

    1. $ curl -H "Accept:application/json" localhost:8083/connectors/ 2. [] (1)
    1 No connectors are currently registered with Kafka Connect.

Deploying the MySQL connector

After starting the Debezium and MySQL services, you are ready to deploy the Debezium MySQL connector so that it can start monitoring the sample MySQL database (inventory).

At this point, you are running the Debezium services, a MySQL database server with a sample inventory database, and the MySQL command line client that is connected to the database. To deploy the MySQL connector, you must:

After the connector is registered, it will start monitoring the database server’s binlog and it will generate change events for each row that changes. - Watch the MySQL connector start

Reviewing the Kafka Connect log output as the connector starts helps you to better understand each task it must complete before it can start monitoring the binlog.

Registering a connector to monitor the inventory database

By registering the Debezium MySQL connector, the connector will start monitoring the MySQL database server’s binlog. The binlog records all of the database’s transactions (such as changes to individual rows and changes to the schemas). When a row in the database changes, Debezium generates a change event.
In a production environment, you would typically either use the Kafka tools to manually create the necessary topics, including specifying the number of replicas, or you’d use the Kafka Connect mechanism for customizing the settings of auto-created topics. However, for this tutorial, Kafka is configured to automatically create the topics with just one replica.

Procedure

  1. Review the configuration of the Debezium MySQL connector that you will register.

Before registering the connector, you should be familiar with its configuration. In the next step, you will register the following connector:

1. { 2. "name": "inventory-connector", (1) 3. "config": { (2) 4. "connector.class": "io.debezium.connector.mysql.MySqlConnector", 5. "tasks.max": "1", (3) 6. "database.hostname": "mysql", (4) 7. "database.port": "3306", 8. "database.user": "debezium", 9. "database.password": "dbz", 10. "database.server.id": "184054", (5) 11. "database.server.name": "dbserver1", (5) 12. "database.include.list": "inventory", (6) 13. "database.history.kafka.bootstrap.servers": "kafka:9092", (7) 14. "database.history.kafka.topic": "schema-changes.inventory" (7) 15. } 16. }
1 The name of the connector.
2 The connector’s configuration.
3 Only one task should operate at any one time. Because the MySQL connector reads the MySQL server’s binlog, using a single connector task ensures proper order and event handling. The Kafka Connect service uses connectors to start one or more tasks that do the work, and it automatically distributes the running tasks across the cluster of Kafka Connect services. If any of the services stop or crash, those tasks will be redistributed to running services.
4 The database host, which is the name of the Docker container running the MySQL server (mysql). Docker manipulates the network stack within the containers so that each linked container can be resolved with /etc/hosts using the container name for the host name. If MySQL were running on a normal network, you would specify the IP address or resolvable host name for this value.
5 A unique server ID and name. The server name is the logical identifier for the MySQL server or cluster of servers. This name will be used as the prefix for all Kafka topics.
6 Only changes in the inventory database will be detected.
7 The connector will store the history of the database schemas in Kafka using this broker (the same broker to which you are sending events) and topic name. Upon restart, the connector will recover the schemas of the database that existed at the point in time in the binlog when the connector should begin reading.

For more information, see MySQL connector configuration properties.
For security reasons, you shouldn’t put passwords or other secrets in plain text into connector configurations. Instead, any secrets should be externalized via the mechanism defined in KIP-297(“Externalizing Secrets for Connect Configurations”).
  1. Open a new terminal, and use the curl command to register the Debezium MySQL connector.

This command uses the Kafka Connect service’s API to submit a POST request against the /connectors resource with a JSON document that describes the new connector (called inventory-connector).

This command uses localhost to connect to the Docker host. If you are using a non-native Docker platform, replace localhost with the IP address of of your Docker host.

1. $ curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "inventory-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "mysql", "database.port": "3306", "database.user": "debezium", "database.password": "dbz", "database.server.id": "184054", "database.server.name": "dbserver1", "database.include.list": "inventory", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.inventory" } }'
Windows users may need to escape the double-quotes. For example: 1. $ curl -i -X POST -H “Accept:application/json” -H “Content-Type:application/json” localhost:8083/connectors/ -d ‘{ \”name\”: \”inventory-connector\”, \”config\”: { \”connector.class\”: \”io.debezium.connector.mysql.MySqlConnector\”, \”tasks.max\”: \”1\”, \”database.hostname\”: \”mysql\”, \”database.port\”: \”3306\”, \”database.user\”: \”debezium\”, \”database.password\”: \”dbz\”, \”database.server.id\”: \”184054\”, \”database.server.name\”: \”dbserver1\”, \”database.include.list\”: \”inventory\”, \”database.history.kafka.bootstrap.servers\”: \”kafka:9092\”, \”database.history.kafka.topic\”: \”dbhistory.inventory\” } }’ Otherwise, you might see an error like the following: 1. {“error_code”:500,”message”:”Unexpected character (‘n’ (code 110)): was expecting double-quote to start field name\n at [Source: (org.glassfish.jersey.message.internal.ReaderInterceptorExecutor$UnCloseableInputStream); line: 1, column: 4]”}
If you use Podman, run the following command: 1. $ curl -i -X POST -H “Accept:application/json” -H “Content-Type:application/json” localhost:8083/connectors/ -d ‘{ “name”: “inventory-connector”, “config”: { “connector.class”: “io.debezium.connector.mysql.MySqlConnector”, “tasks.max”: “1”, “database.hostname”: “0.0.0.0”, “database.port”: “3306”, “database.user”: “debezium”, “database.password”: “dbz”, “database.server.id”: “184054”, “database.server.name”: “dbserver1”, “database.include.list”: “inventory”, “database.history.kafka.bootstrap.servers”: “0.0.0.0:9092”, “database.history.kafka.topic”: “dbhistory.inventory” } }’
  1. Verify that inventory-connector is included in the list of connectors:

1. $ curl -H "Accept:application/json" localhost:8083/connectors/ 2. ["inventory-connector"] 2. Review the connector’s tasks:

1. $ curl -i -X GET -H "Accept:application/json" localhost:8083/connectors/inventory-connector

You should see a response similar to the following (formatted for readability):

``` 1. HTTP/1.1 200 OK 2. Date: Thu, 06 Feb 2020 22:12:03 GMT 3. Content-Type: application/json 4. Content-Length: 531 5. Server: Jetty(9.4.20.v20190813)

  1. {
  2. "name": "inventory-connector",
  3. ...
  4. "tasks": [
  5. {
  6. "connector": "inventory-connector", (1)
  7. "task": 0
  8. }
  9. ]
  10. } ```
1 The connector is running a single task (task 0) to do its work. The connector only supports a single task, because MySQL records all of its activities in one sequential binlog. This means the connector only needs one reader to get a consistent, ordered view of all of the events.

Watching the connector start

When you register a connector, it generates a large amount of log output in the Kafka Connect container. By reviewing this output, you can better understand the process that the connector goes through from the time it is created until it begins reading the MySQL server’s binlog.

After registering the inventory-connector connector, you can review the log output in the Kafka Connect container (connect) to track the connector’s status.

The first few lines show the connector (inventory-connector) being created and started:

1. ...
2. 2021-11-30 01:38:44,223 INFO   ||  [Worker clientId=connect-1, groupId=1] Tasks [inventory-connector-0] configs updated   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
3. 2021-11-30 01:38:44,224 INFO   ||  [Worker clientId=connect-1, groupId=1] Handling task config update by restarting tasks []   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
4. 2021-11-30 01:38:44,224 INFO   ||  [Worker clientId=connect-1, groupId=1] Rebalance started   [org.apache.kafka.connect.runtime.distributed.WorkerCoordinator]
5. 2021-11-30 01:38:44,224 INFO   ||  [Worker clientId=connect-1, groupId=1] (Re-)joining group   [org.apache.kafka.connect.runtime.distributed.WorkerCoordinator]
6. 2021-11-30 01:38:44,227 INFO   ||  [Worker clientId=connect-1, groupId=1] Successfully joined group with generation Generation{generationId=3, memberId='connect-1-7b087c69-8ac5-4c56-9e6b-ec5adabf27e8', protocol='sessioned'}   [org.apache.kafka.connect.runtime.distributed.WorkerCoordinator]
7. 2021-11-30 01:38:44,230 INFO   ||  [Worker clientId=connect-1, groupId=1] Successfully synced group in generation Generation{generationId=3, memberId='connect-1-7b087c69-8ac5-4c56-9e6b-ec5adabf27e8', protocol='sessioned'}   [org.apache.kafka.connect.runtime.distributed.WorkerCoordinator]
8. 2021-11-30 01:38:44,231 INFO   ||  [Worker clientId=connect-1, groupId=1] Joined group at generation 3 with protocol version 2 and got assignment: Assignment{error=0, leader='connect-1-7b087c69-8ac5-4c56-9e6b-ec5adabf27e8', leaderUrl='http://172.17.0.7:8083/', offset=4, connectorIds=[inventory-connector], taskIds=[inventory-connector-0], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with rebalance delay: 0   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
9. 2021-11-30 01:38:44,232 INFO   ||  [Worker clientId=connect-1, groupId=1] Starting connectors and tasks using config offset 4   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
10. 2021-11-30 01:38:44,232 INFO   ||  [Worker clientId=connect-1, groupId=1] Starting task inventory-connector-0   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
11. ...

Further down, you should see output like the following from the connector:

1. ...
2. 2021-11-30 01:38:44,406 INFO   ||  Kafka version: 3.0.0   [org.apache.kafka.common.utils.AppInfoParser]
3. 2021-11-30 01:38:44,406 INFO   ||  Kafka commitId: 8cb0a5e9d3441962   [org.apache.kafka.common.utils.AppInfoParser]
4. 2021-11-30 01:38:44,407 INFO   ||  Kafka startTimeMs: 1638236324406   [org.apache.kafka.common.utils.AppInfoParser]
5. 2021-11-30 01:38:44,437 INFO   ||  Database history topic '(name=dbhistory.inventory, numPartitions=1, replicationFactor=1, replicasAssignments=null, configs={cleanup.policy=delete, retention.ms=9223372036854775807, retention.bytes=-1})' created   [io.debezium.relational.history.KafkaDatabaseHistory]
6. 2021-11-30 01:38:44,497 INFO   ||  App info kafka.admin.client for dbserver1-dbhistory unregistered   [org.apache.kafka.common.utils.AppInfoParser]
7. 2021-11-30 01:38:44,499 INFO   ||  Metrics scheduler closed   [org.apache.kafka.common.metrics.Metrics]
8. 2021-11-30 01:38:44,499 INFO   ||  Closing reporter org.apache.kafka.common.metrics.JmxReporter   [org.apache.kafka.common.metrics.Metrics]
9. 2021-11-30 01:38:44,499 INFO   ||  Metrics reporters closed   [org.apache.kafka.common.metrics.Metrics]
10. 2021-11-30 01:38:44,499 INFO   ||  Reconnecting after finishing schema recovery   [io.debezium.connector.mysql.MySqlConnectorTask]
11. 2021-11-30 01:38:44,524 INFO   ||  Requested thread factory for connector MySqlConnector, id = dbserver1 named = change-event-source-coordinator   [io.debezium.util.Threads]
12. 2021-11-30 01:38:44,525 INFO   ||  Creating thread debezium-mysqlconnector-dbserver1-change-event-source-coordinator   [io.debezium.util.Threads]
13. 2021-11-30 01:38:44,526 INFO   ||  WorkerSourceTask{id=inventory-connector-0} Source task finished initialization and start   [org.apache.kafka.connect.runtime.WorkerSourceTask]
14. 2021-11-30 01:38:44,529 INFO   MySQL|dbserver1|snapshot  Metrics registered   [io.debezium.pipeline.ChangeEventSourceCoordinator]
15. 2021-11-30 01:38:44,529 INFO   MySQL|dbserver1|snapshot  Context created   [io.debezium.pipeline.ChangeEventSourceCoordinator]
16. 2021-11-30 01:38:44,534 INFO   MySQL|dbserver1|snapshot  No previous offset has been found   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
17. 2021-11-30 01:38:44,534 INFO   MySQL|dbserver1|snapshot  According to the connector configuration both schema and data will be snapshotted   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
18. 2021-11-30 01:38:44,534 INFO   MySQL|dbserver1|snapshot  Snapshot step 1 - Preparing   [io.debezium.relational.RelationalSnapshotChangeEventSource]
19. ...

The Debezium log output uses mapped diagnostic contexts (MDC) to provide thread-specific information in the log output, and make it easier to understand what is happening in the multi-threaded Kafka Connect service. This includes the connector type (MySQL in the above log messages), the logical name of the connector (dbserver1 above), and the connector’s activity (task, snapshot and binlog).

In the log output above, the first few lines involve the task activity of the connector, and report some bookkeeping information (in this case, that the connector was started with no prior offset). The next three lines involve the snapshot activity of the connector, and report that a snapshot is being started using the debezium MySQL user along with the MySQL grants associated with that user.
If the connector is not able to connect, or if it does not see any tables or the binlog, check these grants to ensure that all of those listed above are included.

Next, the connector reports the steps that make up the snapshot operation:

1. ...
2. 2021-11-30 01:38:44,534 INFO   MySQL|dbserver1|snapshot  Snapshot step 1 - Preparing   [io.debezium.relational.RelationalSnapshotChangeEventSource]
3. 2021-11-30 01:38:44,535 INFO   MySQL|dbserver1|snapshot  Snapshot step 2 - Determining captured tables   [io.debezium.relational.RelationalSnapshotChangeEventSource]
4. 2021-11-30 01:38:44,535 INFO   MySQL|dbserver1|snapshot  Read list of available databases   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
5. 2021-11-30 01:38:44,537 INFO   MySQL|dbserver1|snapshot       list of available databases is: [information_schema, inventory, mysql, performance_schema, sys]   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
6. 2021-11-30 01:38:44,537 INFO   MySQL|dbserver1|snapshot  Read list of available tables in each database   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
7. 2021-11-30 01:38:44,548 INFO   MySQL|dbserver1|snapshot      snapshot continuing with database(s): [inventory]   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
8. 2021-11-30 01:38:44,551 INFO   MySQL|dbserver1|snapshot  Snapshot step 3 - Locking captured tables [inventory.addresses, inventory.customers, inventory.geom, inventory.orders, inventory.products, inventory.products_on_hand]   [io.debezium.relational.RelationalSnapshotChangeEventSource]
9. 2021-11-30 01:38:44,552 INFO   MySQL|dbserver1|snapshot  Flush and obtain global read lock to prevent writes to database   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
10. 2021-11-30 01:38:44,557 INFO   MySQL|dbserver1|snapshot  Snapshot step 4 - Determining snapshot offset   [io.debezium.relational.RelationalSnapshotChangeEventSource]
11. 2021-11-30 01:38:44,560 INFO   MySQL|dbserver1|snapshot  Read binlog position of MySQL primary server   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
12. 2021-11-30 01:38:44,562 INFO   MySQL|dbserver1|snapshot       using binlog 'mysql-bin.000003' at position '156' and gtid ''   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
13. 2021-11-30 01:38:44,562 INFO   MySQL|dbserver1|snapshot  Snapshot step 5 - Reading structure of captured tables   [io.debezium.relational.RelationalSnapshotChangeEventSource]
14. 2021-11-30 01:38:44,562 INFO   MySQL|dbserver1|snapshot  All eligible tables schema should be captured, capturing: [inventory.addresses, inventory.customers, inventory.geom, inventory.orders, inventory.products, inventory.products_on_hand]   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
15. 2021-11-30 01:38:45,058 INFO   MySQL|dbserver1|snapshot  Reading structure of database 'inventory'   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
16. 2021-11-30 01:38:45,187 INFO   MySQL|dbserver1|snapshot  Snapshot step 6 - Persisting schema history   [io.debezium.relational.RelationalSnapshotChangeEventSource]
17. 2021-11-30 01:38:45,273 INFO   MySQL|dbserver1|snapshot  Releasing global read lock to enable MySQL writes   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
18. 2021-11-30 01:38:45,274 INFO   MySQL|dbserver1|snapshot  Writes to MySQL tables prevented for a total of 00:00:00.717   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
19. 2021-11-30 01:38:45,274 INFO   MySQL|dbserver1|snapshot  Snapshot step 7 - Snapshotting data   [io.debezium.relational.RelationalSnapshotChangeEventSource]
20. 2021-11-30 01:38:45,275 INFO   MySQL|dbserver1|snapshot  Snapshotting contents of 6 tables while still in transaction   [io.debezium.relational.RelationalSnapshotChangeEventSource]
21. 2021-11-30 01:38:45,275 INFO   MySQL|dbserver1|snapshot  Exporting data from table 'inventory.addresses' (1 of 6 tables)   [io.debezium.relational.RelationalSnapshotChangeEventSource]
22. 2021-11-30 01:38:45,276 INFO   MySQL|dbserver1|snapshot       For table 'inventory.addresses' using select statement: 'SELECT `id`, `customer_id`, `street`, `city`, `state`, `zip`, `type` FROM `inventory`.`addresses`'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
23. 2021-11-30 01:38:45,295 INFO   MySQL|dbserver1|snapshot       Finished exporting 7 records for table 'inventory.addresses'; total duration '00:00:00.02'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
24. 2021-11-30 01:38:45,296 INFO   MySQL|dbserver1|snapshot  Exporting data from table 'inventory.customers' (2 of 6 tables)   [io.debezium.relational.RelationalSnapshotChangeEventSource]
25. 2021-11-30 01:38:45,296 INFO   MySQL|dbserver1|snapshot       For table 'inventory.customers' using select statement: 'SELECT `id`, `first_name`, `last_name`, `email` FROM `inventory`.`customers`'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
26. 2021-11-30 01:38:45,304 INFO   MySQL|dbserver1|snapshot       Finished exporting 4 records for table 'inventory.customers'; total duration '00:00:00.008'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
27. 2021-11-30 01:38:45,304 INFO   MySQL|dbserver1|snapshot  Exporting data from table 'inventory.geom' (3 of 6 tables)   [io.debezium.relational.RelationalSnapshotChangeEventSource]
28. 2021-11-30 01:38:45,305 INFO   MySQL|dbserver1|snapshot       For table 'inventory.geom' using select statement: 'SELECT `id`, `g`, `h` FROM `inventory`.`geom`'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
29. 2021-11-30 01:38:45,316 INFO   MySQL|dbserver1|snapshot       Finished exporting 3 records for table 'inventory.geom'; total duration '00:00:00.011'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
30. 2021-11-30 01:38:45,316 INFO   MySQL|dbserver1|snapshot  Exporting data from table 'inventory.orders' (4 of 6 tables)   [io.debezium.relational.RelationalSnapshotChangeEventSource]
31. 2021-11-30 01:38:45,316 INFO   MySQL|dbserver1|snapshot       For table 'inventory.orders' using select statement: 'SELECT `order_number`, `order_date`, `purchaser`, `quantity`, `product_id` FROM `inventory`.`orders`'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
32. 2021-11-30 01:38:45,325 INFO   MySQL|dbserver1|snapshot       Finished exporting 4 records for table 'inventory.orders'; total duration '00:00:00.008'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
33. 2021-11-30 01:38:45,325 INFO   MySQL|dbserver1|snapshot  Exporting data from table 'inventory.products' (5 of 6 tables)   [io.debezium.relational.RelationalSnapshotChangeEventSource]
34. 2021-11-30 01:38:45,325 INFO   MySQL|dbserver1|snapshot       For table 'inventory.products' using select statement: 'SELECT `id`, `name`, `description`, `weight` FROM `inventory`.`products`'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
35. 2021-11-30 01:38:45,343 INFO   MySQL|dbserver1|snapshot       Finished exporting 9 records for table 'inventory.products'; total duration '00:00:00.017'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
36. 2021-11-30 01:38:45,344 INFO   MySQL|dbserver1|snapshot  Exporting data from table 'inventory.products_on_hand' (6 of 6 tables)   [io.debezium.relational.RelationalSnapshotChangeEventSource]
37. 2021-11-30 01:38:45,344 INFO   MySQL|dbserver1|snapshot       For table 'inventory.products_on_hand' using select statement: 'SELECT `product_id`, `quantity` FROM `inventory`.`products_on_hand`'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
38. 2021-11-30 01:38:45,353 INFO   MySQL|dbserver1|snapshot       Finished exporting 9 records for table 'inventory.products_on_hand'; total duration '00:00:00.009'   [io.debezium.relational.RelationalSnapshotChangeEventSource]
39. 2021-11-30 01:38:45,355 INFO   MySQL|dbserver1|snapshot  Snapshot - Final stage   [io.debezium.pipeline.source.AbstractSnapshotChangeEventSource]
40. 2021-11-30 01:38:45,356 INFO   MySQL|dbserver1|snapshot  Snapshot ended with SnapshotResult [status=COMPLETED, offset=MySqlOffsetContext [sourceInfoSchema=Schema{io.debezium.connector.mysql.Source:STRUCT}, sourceInfo=SourceInfo [currentGtid=null, currentBinlogFilename=mysql-bin.000003, currentBinlogPosition=156, currentRowNumber=0, serverId=0, sourceTime=2021-11-30T01:38:45.352Z, threadId=-1, currentQuery=null, tableIds=[inventory.products_on_hand], databaseName=inventory], snapshotCompleted=true, transactionContext=TransactionContext [currentTransactionId=null, perTableEventCount={}, totalEventCount=0], restartGtidSet=null, currentGtidSet=null, restartBinlogFilename=mysql-bin.000003, restartBinlogPosition=156, restartRowsToSkip=0, restartEventsToSkip=0, currentEventLengthInBytes=0, inTransaction=false, transactionId=null, incrementalSnapshotContext =IncrementalSnapshotContext [windowOpened=false, chunkEndPosition=null, dataCollectionsToSnapshot=[], lastEventKeySent=null, maximumKey=null]]]   [io.debezium.pipeline.ChangeEventSourceCoordinator]
41. ...

Each of these steps reports what the connector is doing to perform the consistent snapshot. For example, Step 6 involves reverse engineering the DDL create statements for the tables that are being captured and the global write lock just 1 second after acquiring it, and Step 7 reads all of the rows in each of the tables and reports the time taken and number of rows found. In this case, the connector completed its consistent snapshot in just under 1 second.
The snapshot process will take longer with your databases, but the connector outputs enough log messages that you can track what it is working on, even when the tables have a large number of rows. And although an exclusive write lock is used at the beginning of the snapshot process, it should not last very long even for large databases. This is because the lock is released before any data is copied. For more information, see the MySQL connector documentation.

Next, Kafka Connect reports some “errors”. However, you can safely ignore these warnings: these messages just mean that new Kafka topics were created and that Kafka had to assign a new leader for each one:

``` 1. ... 2. 2021-11-30 01:38:45,555 WARN || [Producer clientId=connector-producer-inventory-connector-0] Error while fetching metadata with correlation id 3 : {dbserver1=LEADER_NOT_AVAILABLE} [org.apache.kafka.clients.NetworkClient] 3. 2021-11-30 01:38:45,691 WARN || [Producer clientId=connector-producer-inventory-connector-0] Error while fetching metadata with correlation id 9 : {dbserver1.inventory.addresses=LEADER_NOT_AVAILABLE} [org.apache.kafka.clients.NetworkClient] 4. 2021-11-30 01:38:45,813 WARN || [Producer clientId=connector-producer-inventory-connector-0] Error while fetching metadata with correlation id 13 : {dbserver1.inventory.customers=LEADER_NOT_AVAILABLE} [org.apache.kafka.clien

（本文超出站点单文档体积上限，已截断）