### Run MySQL Container with Podman

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Starts a MySQL container using Podman. Ensure you are linked to the 'dbz' pod.

```bash
$ 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:3.5
```

--------------------------------

### Verify Kafka Startup

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Checks the output logs to confirm that the Kafka server has started successfully.

```log
...
2025-07-22T11:43:00,935 - INFO  [main:AppInfoParser$AppInfo@125] - Kafka version: 4.0.0
2025-07-22T11:43:00,938 - INFO  [main:AppInfoParser$AppInfo@126] - Kafka commitId: 985bc99521dd22bb
2025-07-22T11:43:00,939 - INFO  [main:AppInfoParser$AppInfo@127] - Kafka startTimeMs: 1753184580923
2025-07-22T11:43:00,959 - INFO  [main:Logging@66] - [KafkaRaftServer nodeId=1] Kafka Server started
```

--------------------------------

### Run Kafka Connect with Podman

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Starts a Kafka Connect container using Podman with similar configurations as the Docker command.

```bash
$ 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:3.5
```

--------------------------------

### Start MySQL Command Line Client with Podman

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Connects to the running MySQL container using the command line client via Podman. Assumes connection to localhost.

```bash
$ podman run -it --rm --name mysqlterm --pod dbz mysql:8.2 sh -c 'exec mysql -h 0.0.0.0 -uroot -pdebezium'
```

--------------------------------

### Run Kafka Container with Podman

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Starts a Kafka container using Podman. Ensure you are linked to the 'dbz' pod.

```bash
$ podman run -it --rm --name kafka --pod dbz -e HOST_NAME=127.0.0.1 quay.io/debezium/kafka:3.5
```

--------------------------------

### Connector Task Status Example

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This is an example of the expected HTTP response when reviewing connector tasks, showing the connector name and its task(s).

```json
{
  "name": "inventory-connector",
  ...
  "tasks": [
    {
      "connector": "inventory-connector",
      "task": 0
    }
  ]
}
```

--------------------------------

### Run MySQL Container with Docker

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Starts a MySQL container with an inventory database using Docker. Ports are mapped for external access.

```bash
$ 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:3.5
```

--------------------------------

### Start MySQL Command Line Client with Docker

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Connects to the running MySQL container using the command line client via Docker. Uses linked container information.

```bash
$ docker run -it --rm --name mysqlterm --link mysql mysql:8.2 sh -c 'exec mysql -h"$MYSQL_PORT_3306_TCP_ADDR" -P"$MYSQL_PORT_3306_TCP_PORT" -umysqluser -p"mysqlpw"'
```

--------------------------------

### Example Kafka Connect Event Key

Source: https://debezium.io/documentation/reference/stable/tutorial.html

An example of the JSON structure for the key of a change event produced by the Debezium MySQL connector. It typically contains the primary key of the database row.

```json
{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1001}}
```

--------------------------------

### Kafka Connect Worker Log Output

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Example log output from the Kafka Connect container showing the connector being created and started, including task configuration updates.

```log
... 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]
```

--------------------------------

### Run Kafka Connect with Docker

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Starts a Kafka Connect container with specified configurations and links to Kafka and MySQL services. The container is interactive, will be removed on stop, and maps port 8083.

```bash
$ 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:3.5
```

--------------------------------

### Debezium Connector Operational Logs

Source: https://debezium.io/documentation/reference/stable/tutorial.html

These logs show the Debezium connector initializing, creating necessary Kafka topics for schema history, and starting the snapshot process for the 'dbserver1' MySQL instance.

```log
2021-11-30 01:38:44,406 INFO   ||  Kafka version: 3.0.0   [org.apache.kafka.common.utils.AppInfoParser]
2021-11-30 01:38:44,406 INFO   ||  Kafka commitId: 8cb0a5e9d3441962   [org.apache.kafka.common.utils.AppInfoParser]
2021-11-30 01:38:44,407 INFO   ||  Kafka startTimeMs: 1638236324406   [org.apache.kafka.common.utils.AppInfoParser]
2021-11-30 01:38:44,437 INFO   ||  Database schema history topic '(name=schemahistory.inventory, numPartitions=1, replicationFactor=1, replicasAssignments=null, configs={cleanup.policy=delete, retention.ms=9223372036854775807, retention.bytes=-1})' created   [io.debezium.storage.kafka.history.KafkaSchemaHistory]
2021-11-30 01:38:44,497 INFO   ||  App info kafka.admin.client for dbserver1-schemahistory unregistered   [org.apache.kafka.common.utils.AppInfoParser]
2021-11-30 01:38:44,499 INFO   ||  Metrics scheduler closed   [org.apache.kafka.common.metrics.Metrics]
2021-11-30 01:38:44,499 INFO   ||  Closing reporter org.apache.kafka.common.metrics.JmxReporter   [org.apache.kafka.common.metrics.Metrics]
2021-11-30 01:38:44,499 INFO   ||  Metrics reporters closed   [org.apache.kafka.common.metrics.Metrics]
2021-11-30 01:38:44,499 INFO   ||  Reconnecting after finishing schema recovery   [io.debezium.connector.mysql.MySqlConnectorTask]
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]
2021-11-30 01:38:44,525 INFO   ||  Creating thread debezium-mysqlconnector-dbserver1-change-event-source-coordinator   [io.debezium.util.Threads]
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]
2021-11-30 01:38:44,529 INFO   MySQL|dbserver1|snapshot  Metrics registered   [io.debezium.pipeline.ChangeEventSourceCoordinator]
2021-11-30 01:38:44,529 INFO   MySQL|dbserver1|snapshot  Context created   [io.debezium.pipeline.ChangeEventSourceCoordinator]
2021-11-30 01:38:44,534 INFO   MySQL|dbserver1|snapshot  No previous offset has been found   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
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]
2021-11-30 01:38:44,534 INFO   MySQL|dbserver1|snapshot  Snapshot step 1 - Preparing   [io.debezium.relational.RelationalSnapshotChangeEventSource]
```

--------------------------------

### Start Kafka Container

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Runs a Kafka container using the Debezium image. Ensure to replace <YOUR_UNIQUE_CLUSTER_IDENTIFIER> with a unique ID. This command maps port 9092 and configures Kafka listeners and quorum voters.

```bash
$ docker run -it --rm -p 9092:9092 \
  --name kafka --hostname kafka \
  -e CLUSTER_ID=<YOUR_UNIQUE_CLUSTER_IDENTIFIER> \
  -e NODE_ID=1 \
  -e NODE_ROLE=combined \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 \
  -e KAFKA_LISTENERS=PLAINTEXT://kafka:9092,CONTROLLER://kafka:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092 \
  quay.io/debezium/kafka:3.5
```

--------------------------------

### Review Connector Tasks

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Retrieve detailed information about the 'inventory-connector', including its tasks, by making a GET request to the Kafka Connect API.

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

--------------------------------

### Debezium Connector Streaming Start Log

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This log message signifies that the Debezium connector has successfully transitioned from the snapshot phase to the continuous streaming of binlog events from the MySQL server.

```log
2021-11-30 01:38:45,362 INFO   MySQL|dbserver1|streaming  Starting streaming   [io.debezium.pipeline.ChangeEventSourceCoordinator]
```

--------------------------------

### Debezium Connector Log Output with MDC

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Example log output demonstrating the use of Mapped Diagnostic Context (MDC) to provide thread-specific information in Debezium logs. This helps in understanding connector type, logical name, and activity like task, snapshot, and binlog.

```log
2021-11-30 01:38:44,534 INFO   MySQL|dbserver1|snapshot  Snapshot step 1 - Preparing   [io.debezium.relational.RelationalSnapshotChangeEventSource]
2021-11-30 01:38:44,535 INFO   MySQL|dbserver1|snapshot  Snapshot step 2 - Determining captured tables   [io.debezium.relational.RelationalSnapshotChangeEventSource]
2021-11-30 01:38:44,535 INFO   MySQL|dbserver1|snapshot  Read list of available databases   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
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]
2021-11-30 01:38:44,537 INFO   MySQL|dbserver1|snapshot  Read list of available tables in each database   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
2021-11-30 01:38:44,548 INFO   MySQL|dbserver1|snapshot  	snapshot continuing with database(s): [inventory]   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
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]
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]
2021-11-30 01:38:44,557 INFO   MySQL|dbserver1|snapshot  Snapshot step 4 - Determining snapshot offset   [io.debezium.relational.RelationalSnapshotChangeEventSource]
2021-11-30 01:38:44,560 INFO   MySQL|dbserver1|snapshot  Read binlog position of MySQL primary server   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
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]
2021-11-30 01:38:44,562 INFO   MySQL|dbserver1|snapshot  Snapshot step 5 - Reading structure of captured tables   [io.debezium.relational.RelationalSnapshotChangeEventSource]
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]
2021-11-30 01:38:45,058 INFO   MySQL|dbserver1|snapshot  Reading structure of database 'inventory'   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
2021-11-30 01:38:45,187 INFO   MySQL|dbserver1|snapshot  Snapshot step 6 - Persisting schema history   [io.debezium.relational.RelationalSnapshotChangeEventSource]
2021-11-30 01:38:45,273 INFO   MySQL|dbserver1|snapshot  Releasing global read lock to enable MySQL writes   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
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]
2021-11-30 01:38:45,274 INFO   MySQL|dbserver1|snapshot  Snapshot step 7 - Snapshotting data   [io.debezium.relational.RelationalSnapshotChangeEventSource]
2021-11-30 01:38:45,275 INFO   MySQL|dbserver1|snapshot  Snapshotting contents of 6 tables while still in transaction   [io.debezium.relational.RelationalSnapshotChangeEventSource]
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]
```

--------------------------------

### Connector Registration Log Output

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This log output indicates that a Kafka Connect worker has successfully registered and started the 'inventory-connector' and its associated task.

```log
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]
2021-11-30 01:38:44,224 INFO   ||  [Worker clientId=connect-1, groupId=1] Rebalance started   [org.apache.kafka.connect.runtime.distributed.WorkerCoordinator]
2021-11-30 01:38:44,224 INFO   ||  [Worker clientId=connect-1, groupId=1] (Re-)joining group   [org.apache.kafka.connect.runtime.distributed.WorkerCoordinator]
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]
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]
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]
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]
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]
```

--------------------------------

### Debezium Event Structure Example

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This JSON object represents a Debezium event for a row creation in the 'inventory.customers' table. It includes the event schema and payload, detailing the 'before' and 'after' states of the row, along with source information from the database.

```json
{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.customers.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.customers.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": true,
            "field": "version"
          },
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_sec"
          },
          {
            "type": "string",
            "optional": true,
            "field": "gtid"
          },
          {
            "type": "string",
            "optional": false,
            "field": "file"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "pos"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "row"
          },
          {
            "type": "boolean",
            "optional": true,
            "field": "snapshot"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "table"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.mysql.Source",
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_us"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ns"
      }
    ],
    "optional": false,
    "name": "dbserver1.inventory.customers.Envelope",
    "version": 1
  },
  "payload": {
    "before": null,
    "after": {
      "id": 1004,
      "first_name": "Anne",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "source": {
      "version": "3.5.2.Final",
      "name": "dbserver1",
      "server_id": 0,
      "ts_sec": 0,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 154,
      "row": 0,
      "snapshot": true,
      "thread": null,
      "db": "inventory",
      "table": "customers"
    },
    "op": "r",
    "ts_ms": 1486500577691,
    "ts_us": 1486500577691547,
    "ts_ns": 1486500577691547930
  }
}
```

--------------------------------

### Check Kafka Connect service status with curl

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Sends an HTTP GET request to the Kafka Connect REST API to verify its status and retrieve version information.

```bash
$ curl -H "Accept:application/json" localhost:8083/
```

--------------------------------

### Verify MySQL Command Line Client Startup

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Confirms that the MySQL command line client has successfully connected to the server.

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

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

--------------------------------

### Verify MySQL Startup

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Checks the output logs to confirm that the MySQL server is ready for connections.

```log
...
[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.
[System] [MY-011323] [Server] X Plugin ready for connections. Bind-address: '::' port: 33060, socket: /var/run/mysqld/mysqlx.sock
```

--------------------------------

### Switch to the inventory database

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Use this command at the MySQL prompt to select the database for your operations.

```sql
mysql> use inventory;
```

--------------------------------

### Watch Kafka Topic with watch-topic Utility

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Launches the `watch-topic` utility in a Docker container to monitor a specified Kafka topic. Use the `-a` flag to watch all events and `-k` to include the event key in the output. This utility is for exploration, not application consumption.

```bash
$ docker run -it --rm --name watcher --link kafka:kafka quay.io/debezium/kafka:3.5 watch-topic -a -k dbserver1.inventory.customers
```

```bash
$ podman run -it --rm --name watcher --pod dbz quay.io/debezium/kafka:3.5 watch-topic -a -k dbserver1.inventory.customers
```

--------------------------------

### Kafka Connect Service Startup Logs

Source: https://debezium.io/documentation/reference/stable/tutorial.html

These log messages indicate that the Debezium MySQL connector has successfully restarted. It connects to MySQL, identifies the necessary binlog file, and resumes operation without re-snapshotting data if a previous offset is found.

```log
... 
2021-11-30 01:49:07,938 INFO   ||  Get all known binlogs from MySQL   [io.debezium.connector.mysql.MySqlConnection] 
2021-11-30 01:49:07,941 INFO   ||  MySQL has the binlog file 'mysql-bin.000003' required by the connector   [io.debezium.connector.mysql.MySqlConnectorTask] 
2021-11-30 01:49:07,967 INFO   ||  Requested thread factory for connector MySqlConnector, id = dbserver1 named = change-event-source-coordinator   [io.debezium.util.Threads] 
2021-11-30 01:49:07,968 INFO   ||  Creating thread debezium-mysqlconnector-dbserver1-change-event-source-coordinator   [io.debezium.util.Threads] 
2021-11-30 01:49:07,968 INFO   ||  WorkerSourceTask{id=inventory-connector-0} Source task finished initialization and start   [org.apache.kafka.connect.runtime.WorkerSourceTask] 
2021-11-30 01:49:07,971 INFO   MySQL|dbserver1|snapshot  Metrics registered   [io.debezium.pipeline.ChangeEventSourceCoordinator] 
2021-11-30 01:49:07,971 INFO   MySQL|dbserver1|snapshot  Context created   [io.debezium.pipeline.ChangeEventSourceCoordinator] 
2021-11-30 01:49:07,976 INFO   MySQL|dbserver1|snapshot  A previous offset indicating a completed snapshot has been found. Neither schema nor data will be snapshotted.   [io.debezium.connector.mysql.MySqlSnapshotChangeEventSource]
```

--------------------------------

### Check Kafka Connectors

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Use this command to verify the list of connectors currently registered with Kafka Connect. An empty list indicates no connectors are active.

```bash
$ curl -H "Accept:application/json" localhost:8083/connectors/
```

--------------------------------

### Stop Debezium Tutorial Podman Pod

Source: https://debezium.io/documentation/reference/stable/tutorial.html

If using Podman, execute these commands to stop and remove the Debezium tutorial pod. This is an alternative to Docker commands for container management.

```bash
$ podman pod kill dbz
$ podman pod rm dbz
```

--------------------------------

### Register Debezium MySQL Connector (Podman)

Source: https://debezium.io/documentation/reference/stable/tutorial.html

If using Podman, adjust the hostname for Kafka bootstrap servers to '0.0.0.0:9092'. This command registers the connector with Podman compatibility.

```bash
$ 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", "topic.prefix": "dbserver1", "database.include.list": "inventory", "schema.history.internal.kafka.bootstrap.servers": "0.0.0.0:9092", "schema.history.internal.kafka.topic": "schemahistory.inventory" } }'
```

--------------------------------

### List tables in the inventory database

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This command displays all tables within the currently selected database.

```sql
mysql> show tables;
```

--------------------------------

### Verify Docker Container Status

Source: https://debezium.io/documentation/reference/stable/tutorial.html

After stopping containers, run this command to confirm that all Debezium tutorial processes have been terminated and removed. It lists all active and stopped containers.

```bash
$ docker ps -a
```

--------------------------------

### Viewing a Create Event for a New Record

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This snippet shows the Kafka Connect event generated for a newly created record. It includes the 'before' and 'after' states of the record, along with source information from the Debezium connector.

```json
{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1005}}	{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"dbserver1.inventory.customers.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"dbserver1.inventory.customers.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":true,"field":"version"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"server_id"},{"type":"int64","optional":false,"field":"ts_sec"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"boolean","optional":true,"field":"snapshot"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"db"},{"type":"string","optional":true,"field":"table"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"dbserver1.inventory.customers.Envelope","version":1},"payload":{"before":null,"after":{"id":1005,"first_name":"Sarah","last_name":"Thompson","email":"kitt@acme.com"},"source":{"version":"3.5.2.Final","name":"dbserver1","server_id":223344,"ts_sec":1490635153,"gtid":null,"file":"mysql-bin.000003","pos":1046,"row":0,"snapshot":null,"thread":3,"db":"inventory","table":"customers"},"op":"c","ts_ms":1490635181455,"ts_us":1490635181455501,"ts_ns":1490635181455501571}}
```

--------------------------------

### Select all data from the customers table

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This SQL query retrieves all rows and columns from the 'customers' table.

```sql
mysql> SELECT * FROM customers;
```

--------------------------------

### Stop Debezium Tutorial Docker Containers

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Use this command to stop all running Docker containers associated with the Debezium tutorial. The `--rm` flag ensures containers are removed after stopping.

```bash
$ docker stop mysqlterm watcher connect mysql kafka
```

--------------------------------

### MySQL Binary Log Client Connection

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This log indicates that the MySQL BinaryLogClient has successfully established a connection to the MySQL server, ready to stream binlog events.

```log
Nov 30, 2021 1:38:45 AM com.github.shyiko.mysql.binlog.BinaryLogClient connect
INFO: Connected to mysql:3306 at mysql-bin.000003/156 (sid:184054, cid:13)
```

--------------------------------

### Querying Customer Data in MySQL

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This SQL query retrieves all records from the 'customers' table. It is used to compare the data in the database with the events generated by the Debezium connector.

```sql
SELECT * FROM customers;
```

--------------------------------

### Debezium MySQL Connector Configuration

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This JSON object defines the configuration for the Debezium MySQL connector. It specifies connection details, database to monitor, and Kafka topic settings.

```json
{
  "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",
    "topic.prefix": "dbserver1",
    "database.include.list": "inventory",
    "schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
    "schema.history.internal.kafka.topic": "schema-changes.inventory"
  }
}
```

--------------------------------

### View Updated 'customers' Table

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This SQL query displays the current state of the 'customers' table after an update. It helps verify the data modification.

```sql
mysql> SELECT * FROM customers;
+------+------------+-----------+-----------------------+
| id   | first_name | last_name | email                 |
+------+------------+-----------+-----------------------+
| 1001 | Sally      | Thomas    | sally.thomas@acme.com |
| 1002 | George     | Bailey    | gbailey@foobar.com    |
| 1003 | Edward     | Walker    | ed@walker.com         |
| 1004 | Anne Marie | Kretchmar | annek@noanswer.org    |
+------+------------+-----------+-----------------------+
4 rows in set (0.00 sec)
```

--------------------------------

### Viewing a Debezium Create Event

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This JSON represents a Debezium event for a newly created customer record. It includes before and after states, source information, and operation type.

```json
{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1006}}	{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"dbserver1.inventory.customers.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"dbserver1.inventory.customers.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":true,"field":"version"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"server_id"},{"type":"int64","optional":false,"field":"ts_sec"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"boolean","optional":true,"field":"snapshot"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"db"},{"type":"string","optional":true,"field":"table"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"dbserver1.inventory.customers.Envelope","version":1},"payload":{"before":null,"after":{"id":1006,"first_name":"Kenneth","last_name":"Anderson","email":"kander@acme.com"},"source":{"version":"3.5.2.Final","name":"dbserver1","server_id":223344,"ts_sec":1490635160,"gtid":null,"file":"mysql-bin.000003","pos":1356,"row":0,"snapshot":null,"thread":3,"db":"inventory","table":"customers"},"op":"c","ts_ms":1490635181456,"ts_us":1490635181456101,"ts_ns":1490635181456101571}}
```

--------------------------------

### Register Debezium MySQL Connector

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Use this curl command to register the Debezium MySQL connector with Kafka Connect API. Ensure secrets are externalized as per KIP-297.

```bash
$ 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", "topic.prefix": "dbserver1", "database.include.list": "inventory", "schema.history.internal.kafka.bootstrap.servers": "kafka:9092", "schema.history.internal.kafka.topic": "schemahistory.inventory" } }'
```

--------------------------------

### MySQL Connector Snapshot Completion Log

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This log entry indicates the successful completion of a consistent snapshot for the 'inventory' database. It details the status, offset information, and the time taken for the snapshot.

```log
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]
```

--------------------------------

### Register Debezium MySQL Connector (Windows Escaped Quotes)

Source: https://debezium.io/documentation/reference/stable/tutorial.html

On Windows, double-quotes within the JSON payload may need to be escaped. This command shows the correctly escaped version for Windows.

```bash
$ 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\", \"topic.prefix\": \"dbserver1\", \"database.include.list\": \"inventory\", \"schema.history.internal.kafka.bootstrap.servers\": \"kafka:9092\", \"schema.history.internal.kafka.topic\": \"schemahistory.inventory\" } }'
```

--------------------------------

### Insert Records into MySQL Database

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Add new records to the 'customers' table in your MySQL database. These inserts will not be captured by Debezium while the Kafka Connect service is stopped.

```sql
mysql> INSERT INTO customers VALUES (default, "Sarah", "Thompson", "kitt@acme.com");
mysql> INSERT INTO customers VALUES (default, "Kenneth", "Anderson", "kander@acme.com");
```

--------------------------------

### Customer Event Key Structure

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This JSON structure represents the key of a Debezium event for the 'customers' table in the 'inventory' database. It includes schema details and the actual payload, which in this case is a customer ID.

```json
{
  "schema":{
    "type":"struct",
      "fields":[
        {
          "type":"int32",
          "optional":false,
          "field":"id"
        }
      ],
    "optional":false,
    "name":"dbserver1.inventory.customers.Key"
  },
  "payload":{
    "id":1004
  }
}
```

--------------------------------

### Kafka Connect Metadata Warnings

Source: https://debezium.io/documentation/reference/stable/tutorial.html

These warnings from Kafka Connect indicate that new topics were created and leaders were assigned. They can be safely ignored in this context.

```log
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]
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]
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.clients.NetworkClient]
2021-11-30 01:38:45,927 WARN   ||  [Producer clientId=connector-producer-inventory-connector-0] Error while fetching metadata with correlation id 18 : {dbserver1.inventory.geom=LEADER_NOT_AVAILABLE}   [org.apache.kafka.clients.NetworkClient]
2021-11-30 01:38:46,043 WARN   ||  [Producer clientId=connector-producer-inventory-connector-0] Error while fetching metadata with correlation id 22 : {dbserver1.inventory.orders=LEADER_NOT_AVAILABLE}   [org.apache.kafka.clients.NetworkClient]
2021-11-30 01:38:46,153 WARN   ||  [Producer clientId=connector-producer-inventory-connector-0] Error while fetching metadata with correlation id 26 : {dbserver1.inventory.products=LEADER_NOT_AVAILABLE}   [org.apache.kafka.clients.NetworkClient]
2021-11-30 01:38:46,269 WARN   ||  [Producer clientId=connector-producer-inventory-connector-0] Error while fetching metadata with correlation id 31 : {dbserver1.inventory.products_on_hand=LEADER_NOT_AVAILABLE}   [org.apache.kafka.clients.NetworkClient]
```

--------------------------------

### Stop Kafka Connect Service

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Use this command to stop the Kafka Connect service container. Docker will remove the container upon stopping if it was run with the --rm option.

```bash
$ docker stop connect
```

--------------------------------

### Debezium Delete Event Key (Second Instance)

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This JSON represents the key of a subsequent Debezium _delete_ event, which is identical to the previous event's key. It confirms the key structure for deleted records.

```json
{
    "schema": {
      "type": "struct",
      "name": "dbserver1.inventory.customers.Key"
      "optional": false,
      "fields": [
        {
          "field": "id",
          "type": "int32",
          "optional": false
        }
      ]
    },
    "payload": {
      "id": 1004
    }
  }
```

--------------------------------

### Update a Record in MySQL

Source: https://debezium.io/documentation/reference/stable/tutorial.html

This SQL statement updates a customer's first name in the 'customers' table. Ensure you are connected to the MySQL client.

```sql
mysql> UPDATE customers SET first_name='Anne Marie' WHERE id=1004;
Query OK, 1 row affected (0.05 sec)
Rows matched: 1  Changed: 1  Warnings: 0
```

--------------------------------

### Delete Record from MySQL

Source: https://debezium.io/documentation/reference/stable/tutorial.html

Use this SQL statement to delete a specific record from the 'customers' table in your MySQL database. This action will trigger a Debezium event.

```sql
mysql> DELETE FROM customers WHERE id=1004;
Query OK, 1 row affected (0.00 sec)
```

--------------------------------

### Handle Foreign Key Constraint Violation

Source: https://debezium.io/documentation/reference/stable/tutorial.html

If the initial delete statement fails due to a foreign key constraint, execute this SQL statement to remove the corresponding address record from the 'addresses' table before retrying the customer deletion.

```sql
mysql> DELETE FROM addresses WHERE customer_id=1004;
```