はじめに

マイクロサービスに関心を抱いている。

実現方法など調べているうちに、
アプリと別サービス間を疎結合(Transactional Outboxパターンなど)にする手法の１つとして「CDC」というものがあることが分かった。

その中の方法に「Debezium」があり、知識としては何ができるか大体わかってるつもりだが実際に触ったことなかったので、この機会に触れてみようと思う。

実施内容

事前準備(環境構築)など非常に面倒に感じる性格なため、
まずは公式(?)が用意している「debezium-examples」を導入して手軽にはじめる。
軽く動かしてみて挙動などが何となく分かったら、
DBサーバを別途構築し[debezium-examples]のDBサーバの変更イベントより、別DBサーバへデータ同期をしてみる。
余力があれば（運用観点で現実的にありそうな）
「OpenSearch」を別途構築し、こちらにもデータ同期を行い、検索＆表示ができることの確認をしたい。

Debeziumについて

Kafkaを軸にしてCDCを実現してくれているようだ。
詳細はググるとたくさん良記事が出てくるので、そちらを参照ください。（TBD:余力あれば追記する）

「debezium-examples」導入

「debezium-examples」のgitクローンをする。

# git clone https://github.com/debezium/debezium-examples.git
# cd debezium-examples/tutorial

コンテナ起動

今回はチュートリアルに記載されている「Using MySQL」にてデモ環境を構築してみる。

DEBEZIUM_VERSIONには、
Dockerイメージ「quay.io/debezium/(zookeeper|kafka|example-mysql)」の最新バージョンを指定する。

「3.1.1.Final」が最新のようだ。

# export DEBEZIUM_VERSION=3.1.1.Final

コンテナ起動前にdocker-compose定義を少々変更

- ネットワーク設定をlinksではなくnetworksに変更

私の場合はMacOS(M3)だったため、docker-compose.yamlに「platform」を指定する必要があった。※Java起因(一例)などでコンテナが異常終了してしまった

diff --git a/tutorial/docker-compose-mysql.yaml b/tutorial/docker-compose-mysql.yaml
index 4be3508..4228ea2 100644
--- a/tutorial/docker-compose-mysql.yaml
+++ b/tutorial/docker-compose-mysql.yaml
@@ -2,28 +2,38 @@ version: '2'
 services:
   zookeeper:
     image: quay.io/debezium/zookeeper:${DEBEZIUM_VERSION}
+    platform: linux/x86_64
     ports:
      - 2181:2181
      - 2888:2888
      - 3888:3888
+    networks:
+      - kafka-demo-net
   kafka:
     image: quay.io/debezium/kafka:${DEBEZIUM_VERSION}
+    platform: linux/x86_64
     ports:
      - 9092:9092
     links:
      - zookeeper
     environment:
      - ZOOKEEPER_CONNECT=zookeeper:2181
+    networks:
+      - kafka-demo-net
   mysql:
     image: quay.io/debezium/example-mysql:${DEBEZIUM_VERSION}
+    platform: linux/x86_64
     ports:
      - 3306:3306
     environment:
      - MYSQL_ROOT_PASSWORD=debezium
      - MYSQL_USER=mysqluser
      - MYSQL_PASSWORD=mysqlpw
+    networks:
+      - kafka-demo-net
   connect:
     image: quay.io/debezium/connect:${DEBEZIUM_VERSION}
+    platform: linux/x86_64
     ports:
      - 8083:8083
     links:
@@ -35,3 +45,8 @@ services:
      - CONFIG_STORAGE_TOPIC=my_connect_configs
      - OFFSET_STORAGE_TOPIC=my_connect_offsets
      - STATUS_STORAGE_TOPIC=my_connect_statuses
+    networks:
+      - kafka-demo-net
+networks:
+  kafka-demo-net:
+    driver: bridge

# docker-compose -f docker-compose-mysql.yaml up

docker-compose起動後のコンテナ一覧は下記の通り。

CONTAINER ID   IMAGE                                        COMMAND                   CREATED          STATUS         PORTS                                                                    NAMES
5321ba94db8a   quay.io/debezium/connect:3.1.1.Final         "/docker-entrypoint.…"   36 seconds ago   Up 35 seconds   8778/tcp, 0.0.0.0:8083->8083/tcp, 9092/tcp                               tutorial-connect-1
6b874f3f92c1   quay.io/debezium/kafka:3.1.1.Final           "/docker-entrypoint.…"   20 minutes ago   Up 35 seconds   0.0.0.0:9092->9092/tcp                                                   tutorial-kafka-1
9309f284de47   quay.io/debezium/example-mysql:3.1.1.Final   "/entrypoint.sh mysq…"   20 minutes ago   Up 35 seconds   0.0.0.0:3306->3306/tcp, 33060-33061/tcp                                  tutorial-mysql-1
8f5ee8b0ffbd   quay.io/debezium/zookeeper:3.1.1.Final       "/docker-entrypoint.…"   20 minutes ago   Up 35 seconds   0.0.0.0:2181->2181/tcp, 0.0.0.0:2888->2888/tcp, 0.0.0.0:3888->3888/tcp   tutorial-zookeeper-1

Mysqlサーバ確認

Mysqlサーバへログイン

 # docker-compose -f docker-compose-mysql.yaml exec mysql bash -c 'mysql -u $MYSQL_USER -p$MYSQL_PASSWORD inventory'

データベース一覧確認

mysql> SHOW DATABASES;
+--------------------+
| Database           |
+--------------------+
| information_schema |
| inventory          | ←変更チャプチャ対象のデータベース（後述）
| performance_schema |
+--------------------+
3 rows in set (0.00 sec)

テーブル一覧

mysql> USE inventory;
Database changed
mysql> 
mysql> 
mysql> SHOW TABLES;
+---------------------+
| Tables_in_inventory |
+---------------------+
| addresses           |
| customers           |
| geom                |
| orders              |
| products            |
| products_on_hand    |
+---------------------+
6 rows in set (0.01 sec)

レコード数の確認。初期レコードは登録されている状態のようだ。

テーブル名	件数
addresses	7
customers	4
geom	3
orders	4
products	9
products_on_hand	9

mysql> select * from addresses;
+----+-------------+---------------------------+------------+--------------+-------+----------+
| id | customer_id | street                    | city       | state        | zip   | type     |
+----+-------------+---------------------------+------------+--------------+-------+----------+
| 10 |        1001 | 3183 Moore Avenue         | Euless     | Texas        | 76036 | SHIPPING |
| 11 |        1001 | 2389 Hidden Valley Road   | Harrisburg | Pennsylvania | 17116 | BILLING  |
| 12 |        1002 | 281 Riverside Drive       | Augusta    | Georgia      | 30901 | BILLING  |
| 13 |        1003 | 3787 Brownton Road        | Columbus   | Mississippi  | 39701 | SHIPPING |
| 14 |        1003 | 2458 Lost Creek Road      | Bethlehem  | Pennsylvania | 18018 | SHIPPING |
| 15 |        1003 | 4800 Simpson Square       | Hillsdale  | Oklahoma     | 73743 | BILLING  |
| 16 |        1004 | 1289 University Hill Road | Canehill   | Arkansas     | 72717 | LIVING   |
+----+-------------+---------------------------+------------+--------------+-------+----------+
7 rows in set (0.01 sec)

mysql> 
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       | Kretchmar | annek@noanswer.org    |
+------+------------+-----------+-----------------------+
4 rows in set (0.00 sec)

mysql> select * from geom;
+----+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+
| id | g                                                                                                                                                                                                    | h          |
+----+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+
|  1 | 0x000000000101000000000000000000F03F000000000000F03F                                                                                                                                                 | NULL       |
|  2 | 0x000000000102000000020000000000000000000040000000000000F03F00000000000018400000000000001840                                                                                                         | NULL       |
|  3 | 0x0000000001030000000100000005000000000000000000000000000000000014400000000000000040000000000000144000000000000000400000000000001C4000000000000000000000000000001C4000000000000000000000000000001440 | NULL       |
+----+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+
3 rows in set (0.01 sec)

mysql> select * from orders;
+--------------+------------+-----------+----------+------------+
| order_number | order_date | purchaser | quantity | product_id |
+--------------+------------+-----------+----------+------------+
|        10001 | 2016-01-16 |      1001 |        1 |        102 |
|        10002 | 2016-01-17 |      1002 |        2 |        105 |
|        10003 | 2016-02-19 |      1002 |        2 |        106 |
|        10004 | 2016-02-21 |      1003 |        1 |        107 |
+--------------+------------+-----------+----------+------------+
4 rows in set (0.00 sec)

mysql> select * from products;
+-----+--------------------+---------------------------------------------------------+--------+
| id  | name               | description                                             | weight |
+-----+--------------------+---------------------------------------------------------+--------+
| 101 | scooter            | Small 2-wheel scooter                                   |   3.14 |
| 102 | car battery        | 12V car battery                                         |    8.1 |
| 103 | 12-pack drill bits | 12-pack of drill bits with sizes ranging from #40 to #3 |    0.8 |
| 104 | hammer             | 12oz carpenter's hammer                                 |   0.75 |
| 105 | hammer             | 14oz carpenter's hammer                                 |  0.875 |
| 106 | hammer             | 16oz carpenter's hammer                                 |      1 |
| 107 | rocks              | box of assorted rocks                                   |    5.3 |
| 108 | jacket             | water resistent black wind breaker                      |    0.1 |
| 109 | spare tire         | 24 inch spare tire                                      |   22.2 |
+-----+--------------------+---------------------------------------------------------+--------+
9 rows in set (0.00 sec)

mysql> select * from products_on_hand;
+------------+----------+
| product_id | quantity |
+------------+----------+
|        101 |        3 |
|        102 |        8 |
|        103 |       18 |
|        104 |        4 |
|        105 |        5 |
|        106 |        0 |
|        107 |       44 |
|        108 |        2 |
|        109 |        5 |
+------------+----------+
9 rows in set (0.01 sec)

動作確認

Kafkaコネクタ登録

チュートリアルにある「Start MySQL connector」を実行してみよう。ただその前に...

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-mysql.json

上記は一体何を（目的は？）しているのだろうか。

REST APIのエンドポイントから、kafka-connectに対して何か命令しているようだ。

新しいコネクタを作成し、成功した場合は現在のコネクタ情報を返します。リバランス処理中の場合、またはコネクタが既に存在する場合は、を返します。409 (Conflict)

コネクタの登録内容は、チュートリアル用の「register-mysql.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"
    }
}

主に、チュートリアル用のmysqlサーバの設定情報が定義されている。
コネクタには「io.debezium.connector.mysql.MySqlConnector(https://debezium.io/documentation/reference/stable/connectors/mysql.html)」が指定されており、プロパティなどコネクタごとに異なっているものと推測する。(公式ドキュメントを見れば何をセットすべきか書いている)

チュートリアルの定義を要約すると、
Mysqlサーバ(ホスト:mysql、ポート：3306、rootユーザー/rootパスワード：debezium/dbz)のデータベース「inventory」のみ変更をキャプチャする、になりそうだ。

コネクタプラグインについて

kafka-connectは下記機能を提供しているとある。

ざっくり、DBとKafkaトピックと紐付けするための命令になっていそうだ。

起動中の Connector プラグインを調べる。

# curl http://localhost:8083/connector-plugins | jq
[
  {
    "class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "type": "sink",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.mongodb.MongoDbSinkConnector",
    "type": "sink",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.db2.Db2Connector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.db2as400.As400RpcConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.informix.InformixConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.mariadb.MariaDbConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.mongodb.MongoDbConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.mysql.MySqlConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.oracle.OracleConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.postgresql.PostgresConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.spanner.SpannerConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.sqlserver.SqlServerConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "io.debezium.connector.vitess.VitessConnector",
    "type": "source",
    "version": "3.1.1.Final"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
    "type": "source",
    "version": "3.9.0"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
    "type": "source",
    "version": "3.9.0"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "type": "source",
    "version": "3.9.0"
  }
]

アクティブなコネクタのリストを確認する。

# curl http://localhost:8083/connectors
[]

まだ何も登録してないので空で応答される。

では、戻って「Start MySQL connector」を実行してみよう。

# curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-mysql.json

HTTP/1.1 201 Created
Date: Sun, 04 May 2025 06:47:27 GMT
Location: http://localhost:8083/connectors/inventory-connector
Content-Type: application/json
Content-Length: 501
Server: Jetty(9.4.56.v20240826)

{"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","name":"inventory-connector"},"tasks":[],"type":"source"}%

登録は成功したようだ。アクティブなコネクタに登録したコネクタ名が表示された。

# curl http://localhost:8083/connectors                                                                                                  
["inventory-connector"]

Debeziumトピックからメッセージを消費

チュートリアルの下記コマンドにて変更キャプチャを確認できそうだ。
一度実行してみよう。

# docker-compose -f docker-compose-mysql.yaml exec kafka /kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server kafka:9092 \
    --from-beginning \
    --property print.key=true \
    --topic dbserver1.inventory.customers


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

  ..(省略)
}

出力が長すぎたので省略した。。
以下属性が配列で返される構造のようだ。

[
  "payload",
  "schema"
]
[
  "payload",
  "schema"
]

「--from-beginning」は、offsetの最初から読み込むためのオプションのため外すとリアルタイムな変更のみにできそうだ。

「--topic」には現在何を指定できるのだろう。何がトピックに用意されているものか確認してみる。
トピック一覧を見るには「/bin/kafka-topics.sh」を使うのが良さそうだ。

上記で実行した「/bin/kafka-console-consumer.sh」は、kafkaコンテナ内にあるbinを指定して実行していた。おそらく同じ場所に「/bin/kafka-topics.sh」があると推測。

以下実行することでトピックを確認することができた。

# docker-compose -f docker-compose-mysql.yaml exec kafka /kafka/bin/kafka-topics.sh \
  --bootstrap-server kafka:9092 \
  --list

dbserver1
dbserver1.inventory.addresses
dbserver1.inventory.customers
dbserver1.inventory.geom
dbserver1.inventory.orders
dbserver1.inventory.products
dbserver1.inventory.products_on_hand
my_connect_configs
my_connect_offsets
my_connect_statuses
schema-changes.inventory

では、実際に「customers」テーブルのレコードを新規追加(insert)した際の変更キャプチャの様子を見てみよう。

変更キャプチャを監視しつつ、

# docker-compose -f docker-compose-mysql.yaml exec kafka /kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server kafka:9092 \
    --property print.key=true \
    --topic dbserver1.inventory.customers

別タブでMysqlサーバへ接続。

# docker-compose -f docker-compose-mysql.yaml exec mysql bash -c 'mysql -u $MYSQL_USER -p$MYSQL_PASSWORD inventory'

mysql> USE inventory;
Database changed

mysql> INSERT INTO inventory.customers(first_name, last_name, email) VALUES('Hoge', 'Fuga1', 'fuga.hoge@demo.org');
Query OK, 1 row affected (0.03 sec)

変更キャプチャに出力あり！

{"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":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,first,first_in_data_collection,last_in_data_collection,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"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":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","version":1,"field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"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":2},"payload":{"before":null,"after":{"id":1005,"first_name":"Hoge","last_name":"Fuga1","email":"fuga.hoge@demo.org"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746344616000,"snapshot":"false","db":"inventory","sequence":null,"ts_us":1746344616000000,"ts_ns":1746344616000000000,"table":"customers","server_id":1,"gtid":null,"file":"binlog.000003","pos":392,"row":0,"thread":18,"query":null},"transaction":null,"op":"c","ts_ms":1746344616957,"ts_us":1746344616957490,"ts_ns":1746344616957490921}}

加工しないと辛い。配列2要素目に有益な情報(before/after)がある。

  "payload": {
    "before": null,
    "after": {
      "id": 1005,
      "first_name": "Hoge",
      "last_name": "Fuga1",
      "email": "fuga.hoge@demo.org"
    },

お次は更新(update)を見てみる。emailを更新。

mysql> UPDATE inventory.customers SET email = 'fuga.hoge.update@demo.org' WHERE id = 1005;

Query OK, 1 row affected (0.02 sec)
Rows matched: 1  Changed: 1  Warnings: 0

変更キャプチャ。

{"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":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,first,first_in_data_collection,last_in_data_collection,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"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":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","version":1,"field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"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":2},"payload":{"before":{"id":1005,"first_name":"Hoge","last_name":"Fuga1","email":"fuga.hoge@demo.org"},"after":{"id":1005,"first_name":"Hoge","last_name":"Fuga1","email":"fuga.hoge.update@demo.org"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746345642000,"snapshot":"false","db":"inventory","sequence":null,"ts_us":1746345642000000,"ts_ns":1746345642000000000,"table":"customers","server_id":1,"gtid":null,"file":"binlog.000003","pos":740,"row":0,"thread":18,"query":null},"transaction":null,"op":"u","ts_ms":1746345642566,"ts_us":1746345642566122,"ts_ns":1746345642566122215}}

更新の場合は、beforeに変更前の情報が入っている。

別DBサーバへのデータ同期

docker-compose-mysql.yamlに同期用Mysqlサーバを追加する。

  mysql-sink:
    image: mysql:8.0
    platform: linux/x86_64
    ports:
      - "3308:3306"
    environment:
      MYSQL_ROOT_PASSWORD: root
      MYSQL_DATABASE: database_b
    volumes:
      - ./mysql-sink-init.sql:/docker-entrypoint-initdb.d/init.sql
    networks:
      - kafka-demo-net

# cat mysql-sink-init.sql
CREATE DATABASE IF NOT EXISTS database_b;
USE database_b;
CREATE TABLE users (
  id INT PRIMARY KEY AUTO_INCREMENT,
  first_name VARCHAR(100),
  last_name VARCHAR(100),
  email VARCHAR(100)
);

# cat register-mysql-sink.json 
{
  "name": "mysql-sink-connector",
  "config": {
    "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:mysql://mysql-sink:3306/database_b",
    "connection.username": "root",
    "connection.password": "root",
    "topics": "dbserver1.inventory.customers",
    "table.name.format": "users",
    "auto.create": "true",
    "auto.evolve": "true",
    "auto.evolve.schemas": "true",
    "insert.mode": "upsert",
    "delete.enabled": "true",
    "primary.key.fields": "id",
    "primary.key.mode": "record_key",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": true,
    "value.converter.schemas.enable": true
  }
}

dockerを再起動

CONTAINER ID   IMAGE                                        COMMAND                   CREATED         STATUS         PORTS                                                                    NAMES
042d24b0f468   mysql:8.0                                    "docker-entrypoint.s…"   3 seconds ago   Up 3 seconds   33060/tcp, 0.0.0.0:3308->3306/tcp                                        tutorial-mysql-sink-1
↑コンテナが追加される

5321ba94db8a   quay.io/debezium/connect:3.1.1.Final         "/docker-entrypoint.…"   2 hours ago     Up 2 hours     8778/tcp, 0.0.0.0:8083->8083/tcp, 9092/tcp                               tutorial-connect-1
6b874f3f92c1   quay.io/debezium/kafka:3.1.1.Final           "/docker-entrypoint.…"   2 hours ago     Up 2 hours     0.0.0.0:9092->9092/tcp                                                   tutorial-kafka-1
9309f284de47   quay.io/debezium/example-mysql:3.1.1.Final   "/entrypoint.sh mysq…"   2 hours ago     Up 2 hours     0.0.0.0:3306->3306/tcp, 33060-33061/tcp                                  tutorial-mysql-1
8f5ee8b0ffbd   quay.io/debezium/zookeeper:3.1.1.Final       "/docker-entrypoint.…"   2 hours ago     Up 2 hours     0.0.0.0:2181->2181/tcp, 0.0.0.0:2888->2888/tcp, 0.0.0.0:3888->3888/tcp   tutorial-zookeeper-1

同期用Mysqlサーバ確認

# docker-compose -f docker-compose-mysql.yaml exec mysql-sink bash -c 'mysql -u root -proot database_b'

mysql> SHOW DATABASES;
+--------------------+
| Database           |
+--------------------+
| database_b         |
| information_schema |
| mysql              |
| performance_schema |
| sys                |
+--------------------+
5 rows in set (0.00 sec)


mysql> USE database_b
Database changed

mysql> SHOW TABLES;
+----------------------+
| Tables_in_database_b |
+----------------------+
| users                |
+----------------------+
1 row in set (0.00 sec)

mysql> select * from users;
Empty set (0.01 sec)

同期用コネクタを追加する。

# curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-mysql-sink.json

HTTP/1.1 201 Created
Date: Sun, 04 May 2025 08:29:57 GMT
Location: http://localhost:8083/connectors/mysql-sink-connector
Content-Type: application/json
Content-Length: 744
Server: Jetty(9.4.56.v20240826)

{"name":"mysql-sink-connector","config":{"connector.class":"io.debezium.connector.jdbc.JdbcSinkConnector","tasks.max":"1","connection.url":"jdbc:mysql://mysql-sink:3306/database_b","connection.username":"root","connection.password":"root","topics":"dbserver1.inventory.customers","table.name.format":"users","auto.create":"true","auto.evolve":"true","auto.evolve.schemas":"true","insert.mode":"upsert","delete.enabled":"true","primary.key.fields":"id","primary.key.mode":"record_key","key.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter":"org.apache.kafka.connect.json.JsonConverter","key.converter.schemas.enable":"true","value.converter.schemas.enable":"true","name":"mysql-sink-connector"},"tasks":[],"type":"sink"}

# curl -i -H "Accept:application/json" http://localhost:8083/connectors/mysql-sink-connector/status 

{"name":"mysql-sink-connector","connector":{"state":"RUNNING","worker_id":"172.20.0.5:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.20.0.5:8083"}],"type":"sink"}

# curl http://localhost:8083/connectors                                          
["mysql-sink-connector","inventory-connector"]

では、新規データを登録して同期されるか見てみよう。

# docker-compose -f docker-compose-mysql.yaml exec mysql bash -c 'mysql -u $MYSQL_USER -p$MYSQL_PASSWORD inventory'

mysql> USE inventory;

mysql> INSERT INTO inventory.customers(first_name, last_name, email) VALUES('Hoge', 'Fuga2', 'fuga2.hoge@demo.org');

Query OK, 1 row affected (0.01 sec)

同期先Mysqlを確認。

# docker-compose -f docker-compose-mysql.yaml exec mysql-sink bash -c 'mysql -u root -proot database_b'

mysql> select * from users;
+------+------------+-----------+---------------------------+
| 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       | Kretchmar | annek@noanswer.org        |
| 1005 | Hoge       | Fuga1     | fuga.hoge.update@demo.org |
| 1006 | Hoge       | Fuga2     | fuga2.hoge@demo.org       |
+------+------------+-----------+---------------------------+
6 rows in set (0.00 sec)

全レコードが登録されていた。
コネクタ追加した時点で追加されていたのだろう...
新規レコードも存在するため同期は成功していそうだ。

更新も試してみよう。先ほど同様にemailを更新する。

# docker-compose -f docker-compose-mysql.yaml exec mysql bash -c 'mysql -u $MYSQL_USER -p$MYSQL_PASSWORD inventory'

mysql> USE inventory;

mysql> UPDATE inventory.customers SET email = 'fuga2.hoge.update@demo.org' WHERE id = 1006;

Query OK, 1 row affected (0.01 sec)
Rows matched: 1  Changed: 1  Warnings: 0

同期先Mysqlを確認。同期されていた。

mysql> select * from users;
+------+------------+-----------+----------------------------+
| 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       | Kretchmar | annek@noanswer.org         |
| 1005 | Hoge       | Fuga1     | fuga.hoge.update@demo.org  |
| 1006 | Hoge       | Fuga2     | fuga2.hoge.update@demo.org |
+------+------------+-----------+----------------------------+
6 rows in set (0.01 sec)

「OpenSearch」へのデータ同期

kafkaコネクタにOpeaSearch同期用プラグインをインストールする必要がある。
https://www.confluent.io/hub/confluentinc/kafka-connect-elasticsearch

「Self-Hosted」のDownloadで取得。

tutorial配下に「connect-plugins」ディレクトリ作成し、ダウンロードしたものを移動する。

# ls -l tutorial/connect-plugins 
total 0
drwxr-xr-x@ 8 tetsu  staff  256  5  4 18:15 confluentinc-kafka-connect-elasticsearch-15.0.0

そしてconnectコンテナのvolumesにてplugin置き場に追加すると使用することができる。

  connect:
    image: quay.io/debezium/connect:${DEBEZIUM_VERSION}
    (..省略)
    volumes:
      - ./connect-plugins/confluentinc-kafka-connect-elasticsearch-15.0.0:/kafka/connect/confluentinc-kafka-connect-elasticsearch-15.0.0

プラグイン確認。

# curl -s localhost:8083/connector-plugins | jq '.[].class' 
"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector"
↑これが追加される。

"io.debezium.connector.jdbc.JdbcSinkConnector"
"io.debezium.connector.mongodb.MongoDbSinkConnector"
"io.debezium.connector.db2.Db2Connector"
"io.debezium.connector.db2as400.As400RpcConnector"
"io.debezium.connector.informix.InformixConnector"
"io.debezium.connector.mariadb.MariaDbConnector"
"io.debezium.connector.mongodb.MongoDbConnector"
"io.debezium.connector.mysql.MySqlConnector"
"io.debezium.connector.oracle.OracleConnector"
"io.debezium.connector.postgresql.PostgresConnector"
"io.debezium.connector.spanner.SpannerConnector"
"io.debezium.connector.sqlserver.SqlServerConnector"
"io.debezium.connector.vitess.VitessConnector"
"org.apache.kafka.connect.mirror.MirrorCheckpointConnector"
"org.apache.kafka.connect.mirror.MirrorHeartbeatConnector"
"org.apache.kafka.connect.mirror.MirrorSourceConnector"

以下コンテナ定義を追加。
→kibanaはデータ検索目的のコンテナなので、kafka連携は不要と判断し「kafka-demo-net」ネットワークには属さない設定にした。

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch-wolfi:8.18.0
    #platform: linux/x86_64
    ports:
      - 9200:9200
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - cluster.name=docker-cluster
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    networks:
      - kafka-demo-net
  kibana:
    image: docker.elastic.co/kibana/kibana:8.18.0
    #platform: linux/x86_64
    environment:
      ELASTICSEARCH_HOSTS: http://elasticsearch:9200
    ports:
       - 5601:5601
    #networks:
    #  - kafka-demo-net

コンテナ再起動。

CONTAINER ID   IMAGE                                                        COMMAND                   CREATED              STATUS              PORTS                                                                    NAMES
ef09ab1655f2   docker.elastic.co/elasticsearch/elasticsearch-wolfi:8.18.0   "/sbin/tini -- /usr/…"   About a minute ago   Up About a minute   0.0.0.0:9200->9200/tcp, 9300/tcp                                         tutorial-elasticsearch-1
ecee9b8e9d3b   docker.elastic.co/kibana/kibana:8.18.0                       "/bin/tini -- /usr/l…"   3 minutes ago        Up 3 minutes        0.0.0.0:5601->5601/tcp                                                   tutorial-kibana-1
↑2コンテナ追加される

042d24b0f468   mysql:8.0                                                    "docker-entrypoint.s…"   43 minutes ago       Up 43 minutes       33060/tcp, 0.0.0.0:3308->3306/tcp                                        tutorial-mysql-sink-1
5321ba94db8a   quay.io/debezium/connect:3.1.1.Final                         "/docker-entrypoint.…"   2 hours ago          Up 2 hours          8778/tcp, 0.0.0.0:8083->8083/tcp, 9092/tcp                               tutorial-connect-1
6b874f3f92c1   quay.io/debezium/kafka:3.1.1.Final                           "/docker-entrypoint.…"   3 hours ago          Up 2 hours          0.0.0.0:9092->9092/tcp                                                   tutorial-kafka-1
9309f284de47   quay.io/debezium/example-mysql:3.1.1.Final                   "/entrypoint.sh mysq…"   3 hours ago          Up 2 hours          0.0.0.0:3306->3306/tcp, 33060-33061/tcp                                  tutorial-mysql-1
8f5ee8b0ffbd   quay.io/debezium/zookeeper:3.1.1.Final                       "/docker-entrypoint.…"   3 hours ago          Up 2 hours          0.0.0.0:2181->2181/tcp, 0.0.0.0:2888->2888/tcp, 0.0.0.0:3888->3888/tcp   tutorial-zookeeper-1

コネクタを追加する。(中身の細かな設定は適当..)

# cat register-opensearch-sink.json
{
  "name": "elasticsearch-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "dbserver1.inventory.customers",
    "key.ignore": "true",
    "schema.ignore": "true",
    "connection.url": "http://elasticsearch:9200",
    "type.name": "_doc",
    "name": "elasticsearch-sink",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": true,
    "value.converter.schemas.enable": true
  }
}

# curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-opensearch-sink.json

{"name":"opensearch-sink-connector","config":{"connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","tasks.max":"1","topics":"dbserver1.inventory.customers","key.ignore":"true","schema.ignore":"true","connection.url":"http://elasticsearch:9200","type.name":"_doc","key.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter":"org.apache.kafka.connect.json.JsonConverter","key.converter.schemas.enable":"true","value.converter.schemas.enable":"true","auto.create":"true","auto.evolve":"true","auto.evolve.schemas":"true","name":"opensearch-sink-connector"},"tasks":[],"type":"sink"}

# curl http://localhost:8083/connectors
["opensearch-sink-connector","mysql-sink-connector","inventory-connector"]

# curl -i -H "Accept:application/json" http://localhost:8083/connectors/opensearch-sink-connector/status

{"name":"opensearch-sink-connector","connector":{"state":"RUNNING","worker_id":"172.20.0.7:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.20.0.7:8083"}],"type":"sink"}

レコード追加。

docker-compose -f docker-compose-mysql.yaml exec mysql bash -c 'mysql -u $MYSQL_USER -p$MYSQL_PASSWORD inventory'


INSERT INTO inventory.customers(first_name, last_name, email) VALUES('Hoge', 'Fuga3', 'fuga3.hoge@demo.org');

OpenSearchのインデックス確認

# curl -XGET 'localhost:9200/_cat/indices?v'         
health status index                         uuid                   pri rep docs.count docs.deleted store.size pri.store.size dataset.size
yellow open   dbserver1.inventory.customers UOH6a_YPRienxo8TtWI6_A   1   1          8            0     20.9kb         20.9kb       20.9kb

「dbserver1.inventory.customers」インデックスが登録されている。

curl -XGET 'localhost:9200/dbserver1.inventory.customers/_search'

{"took":80,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":8,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"dbserver1.inventory.customers","_id":"dbserver1.inventory.customers+0+0","_score":1.0,"_source":{"before":null,"after":{"id":1001,"first_name":"Sally","last_name":"Thomas","email":"sally.thomas@acme.com"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746341249000,"snapshot":"first_in_data_collection","db":"inventory","sequence":null,"ts_us":1746341249000000,"ts_ns":1746341249000000000,"table":"customers","server_id":0,"gtid":null,"file":"binlog.000003","pos":157,"row":0,"thread":null,"query":null},"transaction":null,"op":"r","ts_ms":1746341249993,"ts_us":1746341249993074,"ts_ns":1746341249993074126}},{"_index":"dbserver1.inventory.customers","_id":"dbserver1.inventory.customers+0+1","_score":1.0,"_source":{"before":null,"after":{"id":1002,"first_name":"George","last_name":"Bailey","email":"gbailey@foobar.com"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746341249000,"snapshot":"true","db":"inventory","sequence":null,"ts_us":1746341249000000,"ts_ns":1746341249000000000,"table":"customers","server_id":0,"gtid":null,"file":"binlog.000003","pos":157,"row":0,"thread":null,"query":null},"transaction":null,"op":"r","ts_ms":1746341249993,"ts_us":1746341249993500,"ts_ns":1746341249993500501}},{"_index":"dbserver1.inventory.customers","_id":"dbserver1.inventory.customers+0+2","_score":1.0,"_source":{"before":null,"after":{"id":1003,"first_name":"Edward","last_name":"Walker","email":"ed@walker.com"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746341249000,"snapshot":"true","db":"inventory","sequence":null,"ts_us":1746341249000000,"ts_ns":1746341249000000000,"table":"customers","server_id":0,"gtid":null,"file":"binlog.000003","pos":157,"row":0,"thread":null,"query":null},"transaction":null,"op":"r","ts_ms":1746341249994,"ts_us":1746341249994315,"ts_ns":1746341249994315793}},{"_index":"dbserver1.inventory.customers","_id":"dbserver1.inventory.customers+0+3","_score":1.0,"_source":{"before":null,"after":{"id":1004,"first_name":"Anne","last_name":"Kretchmar","email":"annek@noanswer.org"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746341249000,"snapshot":"last_in_data_collection","db":"inventory","sequence":null,"ts_us":1746341249000000,"ts_ns":1746341249000000000,"table":"customers","server_id":0,"gtid":null,"file":"binlog.000003","pos":157,"row":0,"thread":null,"query":null},"transaction":null,"op":"r","ts_ms":1746341249994,"ts_us":1746341249994462,"ts_ns":1746341249994462918}},{"_index":"dbserver1.inventory.customers","_id":"dbserver1.inventory.customers+0+4","_score":1.0,"_source":{"before":null,"after":{"id":1005,"first_name":"Hoge","last_name":"Fuga1","email":"fuga.hoge@demo.org"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746344616000,"snapshot":"false","db":"inventory","sequence":null,"ts_us":1746344616000000,"ts_ns":1746344616000000000,"table":"customers","server_id":1,"gtid":null,"file":"binlog.000003","pos":392,"row":0,"thread":18,"query":null},"transaction":null,"op":"c","ts_ms":1746344616957,"ts_us":1746344616957490,"ts_ns":1746344616957490921}},{"_index":"dbserver1.inventory.customers","_id":"dbserver1.inventory.customers+0+6","_score":1.0,"_source":{"before":null,"after":{"id":1006,"first_name":"Hoge","last_name":"Fuga2","email":"fuga2.hoge@demo.org"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746347664000,"snapshot":"false","db":"inventory","sequence":null,"ts_us":1746347664000000,"ts_ns":1746347664000000000,"table":"customers","server_id":1,"gtid":null,"file":"binlog.000003","pos":1125,"row":0,"thread":20,"query":null},"transaction":null,"op":"c","ts_ms":1746347664616,"ts_us":1746347664616974,"ts_ns":1746347664616974554}},{"_index":"dbserver1.inventory.customers","_id":"dbserver1.inventory.customers+0+5","_score":1.0,"_source":{"before":{"id":1005,"first_name":"Hoge","last_name":"Fuga1","email":"fuga.hoge@demo.org"},"after":{"id":1005,"first_name":"Hoge","last_name":"Fuga1","email":"fuga.hoge.update@demo.org"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746345642000,"snapshot":"false","db":"inventory","sequence":null,"ts_us":1746345642000000,"ts_ns":1746345642000000000,"table":"customers","server_id":1,"gtid":null,"file":"binlog.000003","pos":740,"row":0,"thread":18,"query":null},"transaction":null,"op":"u","ts_ms":1746345642566,"ts_us":1746345642566122,"ts_ns":1746345642566122215}},{"_index":"dbserver1.inventory.customers","_id":"dbserver1.inventory.customers+0+7","_score":1.0,"_source":{"before":{"id":1006,"first_name":"Hoge","last_name":"Fuga2","email":"fuga2.hoge@demo.org"},"after":{"id":1006,"first_name":"Hoge","last_name":"Fuga2","email":"fuga2.hoge.update@demo.org"},"source":{"version":"3.1.1.Final","connector":"mysql","name":"dbserver1","ts_ms":1746347991000,"snapshot":"false","db":"inventory","sequence":null,"ts_us":1746347991000000,"ts_ns":1746347991000000000,"table":"customers","server_id":1,"gtid":null,"file":"binlog.000003","pos":1474,"row":0,"thread":20,"query":null},"transaction":null,"op":"u","ts_ms":1746347991207,"ts_us":1746347991207050,"ts_ns":1746347991207050969}}]}}

うーん、
customesのレコードが単に連携されることを期待していたが、変更キャプチャ自体が連携されている。

一旦、ここまで。