Debezium - データソースの変更を検知して Kafka Topics に格納されたメッセージを確認する

# 概要
マイクロサービス化案件に携わっており、Debezium を触ったので、明日のためにメモしときます。

## 登場人物
### Kafka
オープンソースの分散メッセージングシステム。メッセージキューって言ってよいのかな。
メッセージキューはマイクロサービスにおけるバックエンドの要。

### Kafka Connect
Kafka と他のシステム間でデータをスケーラブルかつ、ストリーミングするためのソフトウェア。
Kafka との接続方法やリバランスなど、手間のかかる問題を処理してくれる。

### Debezium
データの変更をキャプチャするためのオープンソースの分散プラットフォーム。
データソース （DBなど） の変更を監視し、変更内容を Kafka Topics に格納する。 
Kafka 上に構築されている。
KafKa Connect が使用されている。
トランザクションログを監視している。

### ZooKeeper
大規模分散システムでよく利用される、設定情報の集中管理や名前付けなどのサービスを提供するソフトウェア。
kafka の実行に必要なようで。

<img width="800" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/62209/8b265389-9934-1a58-8490-523402a91dc6.png">
引用: [DEBEZIUM アーキテクチャー](https://access.redhat.com/documentation/ja-jp/red_hat_integration/2020-q2/html/debezium_user_guide/debezium-architecture)
## 手順
最初は [Tutorial](https://debezium.io/documentation/reference/tutorial.html) に沿ってやっていきます。  

#### 1. ZooKeeperの起動

```bash
$ docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:1.8
```

#### 2. Kafkaの起動

```bash
$ docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper debezium/kafka:1.8
```

#### 3. MySQLの起動 
サンプルスキーマが含まれているイメージです。

```bash
$ docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:1.8
```

#### 4. MySQLコマンドラインクライアントの起動
まだ使わないけど、起動しておきます。

```bash
$ 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"'
```

#### 5. KafkaConnectを起動
Kafka と MySQL のデータ連携に必要です。
API で操作可能なので curl 使います。

```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 zookeeper:zookeeper --link kafka:kafka --link mysql:mysql debezium/connect:1.8
```

DebeziumMySQLコネクタを登録します。

```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", "database.server.name": "dbserver1", "database.include.list": "inventory", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.inventory" } }'
```

ここまでで、画像のように起動できていればOK
<img width="400" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/62209/af396d1c-70da-7a49-a979-ec0253bffbdb.png">

#### 6. 登録されたメッセージを確認する
Kafka 実行中のコンテナに入る。

```bash
$ docker exec -it kafka bash
```

以下のコマンドを使用すると、Kafka に入ってきたメッセージの内容をリアルタイムに確認できます。

```bash
[kafka@851e7f6e1d62 ~]$ bin/kafka-console-consumer.sh --topic dbserver1.inventory.customers --bootstrap-server kafka:9092
```

[--topic dbserver1.inventory.customers]
以下のコマンドで確認。
`$ curl -i -X GET -H "Accept:application/json" localhost:8083/connectors/inventory-connector/topics`

[--from-beginning] 
今回使用していないが、このパラメーターを使用することで、全メッセージを確認できる。

#### 7. データに変更を加える
MySQLコマンドラインクライアントで

```sql
use inventory;
UPDATE customers SET first_name='Anne123' WHERE id=1004;
```

customers テーブルへ変更を加えると、`kafka-console-consumer.sh` の画面に変更内容のメッセージキューが入ってくることが確認できる。

``` サンプルメッセージ
{
  "schema": {
    "type": "struct",
    "fields": [
      {
          // 省略...
      }
  },
  "payload": {
    "before": {
      "id": 1004,
      "first_name": "Anne", // ★変更前
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "after": {
      "id": 1004,
      "first_name": "Anne123", // ★変更後
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "source": {
      "version": "1.8.0.Final",
      "connector": "mysql",
      "name": "dbserver1",
      "ts_ms": 1640265034000,
      "snapshot": "false",
      "db": "inventory",
      "sequence": null,
      "table": "customers",
      "server_id": 223344,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 1599,
      "row": 0,
      "thread": null,
      "query": null
    },
    "op": "u",
    "ts_ms": 1640265034477,
    "transaction": null
  }
}
```

## 続き

明日は、上記の環境を windows PC で構築し、メッセージキューに入ってきた内容で lambda を叩いてみます。
よく使う環境なら docker-compose にするのが良いんだろうな…

2021-12-24 追記
Windows でうまくコマンドを実行できないときは PowerShell を使うと良い。
Windows には Curl は別途インストールが必要。（または Postman などの API Client）