OpenShift上に構成したConfluent PlatformとMySQLをDebeziumのConnectorで連携してみる

# はじめに
OpenShift上に構成したConfluent PlatformとMySQLをDebeziumのMySQL Source Connectorを利用してデータ連携してみます。

# 前提
+ IBM Cloud® Virtual Private Cloud(VPC)にOpenShift Container Platform(OCP) 4.7を構成
+ OCP上にConfluent Platform 6.0.0を構成（Confluent Operator 1.7.0を利用）
+ OCP上にMySQL 8.0を構成

# 参考
[Debezium MySQL connector configuration example](https://debezium.io/documentation/reference/1.6/connectors/mysql.html#mysql-example-configuration)
[Debezium MySQL Source Connector Configuration Properties - Database History Parameters](https://docs.confluent.io/debezium-connect-mysql-source/current/mysql_source_connector_config.html#database-history-parameters)
[MySQLのTable更新情報をKafka上でksql処理する](https://qiita.com/ssqwork/items/bc5ff578b1bc70ff26e1)


# 構成確認

Confluent Platformが正常に起動していることを確認する。

```
>oc get all -n confluent-namespace
I0912 23:39:24.059883   37555 request.go:621] Throttling request took 1.090165589s, request: GET:https://c100-e.jp-tok.containers.cloud.ibm.com:31330/apis/packages.operators.coreos.com/v1?timeout=32s
NAME                               READY   STATUS    RESTARTS   AGE
pod/cc-operator-6ffb5f5489-b6l6r   1/1     Running   0          13h
pod/connectors-0                   1/1     Running   0          13h
pod/connectors-1                   1/1     Running   0          13h
pod/controlcenter-0                1/1     Running   0          13h
pod/kafka-0                        1/1     Running   0          13h
pod/kafka-1                        1/1     Running   0          13h
pod/kafka-2                        1/1     Running   0          13h
pod/ksql-0                         1/1     Running   0          13h
pod/ksql-1                         1/1     Running   0          13h
pod/replicator-0                   1/1     Running   0          13h
pod/replicator-1                   1/1     Running   0          13h
pod/schemaregistry-0               1/1     Running   0          13h
pod/zookeeper-0                    1/1     Running   0          13h
pod/zookeeper-1                    1/1     Running   0          13h
pod/zookeeper-2                    1/1     Running   0          13h
(以下、省略)
```
※Kafkaサーバーのauto.create.topics.enableプロパティはtrueにしておくこと。

MySQLが正常に稼働していることを確認する。

```
>oc get all -n shoheim

I0912 23:02:28.265544   36803 request.go:621] Throttling request took 1.148324247s, request: GET:https://c100-e.jp-tok.containers.cloud.ibm.com:31330/apis/monitoring.operator.ibm.com/v1alpha1?timeout=32s
NAME                 READY   STATUS      RESTARTS   AGE
pod/mysql-1-deploy   0/1     Completed   0          45s
pod/mysql-1-z54sl    1/1     Running     0          38s
(以下、省略)
```
※ユーザー権限やログ周りの設定は以下を参考に実施しておく。
[Debezium connector for MySQL - Set up](https://debezium.io/documentation/reference/1.6/connectors/mysql.html#setting-up-mysql)

MySQLには以下のデータベース、テーブルを構成しておく。

```
mysql> show databases;
+--------------------+
| Database           |
+--------------------+
| information_schema |
| opdb               |
+--------------------+

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

mysql> SELECT * from menus;
+------+-----------+
| id   | name      |
+------+-----------+
|    1 | curry     |
|    2 | hamburger |
+------+-----------+
2 rows in set (0.00 sec)
```

# MySQL Source Connectorの構成

## 1. MySQL Connector Pluginのインストール

Confluent Hubクライアント(confluent-hubコマンド)を利用して、ConnectorsのポッドにDebezium MySQL ConnectorのPluginをインストールする。


```
>oc exec -it connectors-0 /bin/bash
bash-4.4$ confluent-hub install debezium/debezium-connector-mysql:1.6.0
The component can be installed in any of the following Confluent Platform installations: 
  1. / (installed rpm/deb package) 
  2. / (where this tool is installed) 
Choose one of these to continue the installation (1-2): 2
Do you want to install this into /usr/share/confluent-hub-components? (yN) y

 
Component's license: 
Apache 2.0 
https://github.com/debezium/debezium/blob/master/LICENSE.txt 
I agree to the software license agreement (yN) y

You are about to install 'debezium-connector-mysql' from Debezium Community, as published on Confluent Hub. 
Do you want to continue? (yN) y

Downloading component Debezium MySQL CDC Connector 1.6.0, provided by Debezium Community from Confluent Hub and installing into /usr/share/confluent-hub-components 
Detected Worker's configs: 
  1. Standard: /etc/kafka/connect-distributed.properties 
  2. Standard: /etc/kafka/connect-standalone.properties 
  3. Standard: /etc/schema-registry/connect-avro-distributed.properties 
  4. Standard: /etc/schema-registry/connect-avro-standalone.properties 
  5. Used by Connect process with PID : /opt/confluent/etc/connect/connect.properties 
Do you want to update all detected configs? (yN) y

Adding installation directory to plugin path in the following files: 
  /etc/kafka/connect-distributed.properties 
  /etc/kafka/connect-standalone.properties 
  /etc/schema-registry/connect-avro-distributed.properties 
  /etc/schema-registry/connect-avro-standalone.properties 
  /opt/confluent/etc/connect/connect.properties 
 
Completed 
```

※ここでは、Pluginのインストール先を/usr/share/confluent-hub-componentsとしている。そのため、事前にConnectorsポッドには上記ディレクトリをマウントポイントとするPersistent Volumeを構成している。

インストールされたものを確認

```
bash-4.4$ cd /usr/share/confluent-hub-components
bash-4.4$ ls -ltr
total 0
drwxrwxrwx. 5 1001 root 63 Oct 25 02:26 debezium-debezium-connector-mysql
bash-4.4$ ls -l debezium-debezium-connector-mysql
total 8
drwxrwxrwx. 2 1001 root   59 Oct 25 02:26 assets
drwxrwxrwx. 2 1001 root  175 Oct 25 02:26 doc
drwxrwxrwx. 2 1001 root 4096 Oct 25 02:26 lib
-rw-rw-rw-. 1 1001 root 2868 Oct 25 02:26 manifest.json
bash-4.4$ ls -l debezium-debezium-connector-mysql/lib
total 9624
-rw-rw-rw-. 1 1001 root  337864 Oct 25 02:26 antlr4-runtime-4.8.jar
-rw-rw-rw-. 1 1001 root   20743 Oct 25 02:26 debezium-api-1.6.0.Final.jar
-rw-rw-rw-. 1 1001 root  376353 Oct 25 02:26 debezium-connector-mysql-1.6.0.Final.jar
-rw-rw-rw-. 1 1001 root  866648 Oct 25 02:26 debezium-core-1.6.0.Final.jar
-rw-rw-rw-. 1 1001 root 2777855 Oct 25 02:26 debezium-ddl-parser-1.6.0.Final.jar
-rw-rw-rw-. 1 1001 root    4617 Oct 25 02:26 failureaccess-1.0.1.jar
-rw-rw-rw-. 1 1001 root 2858426 Oct 25 02:26 guava-30.0-jre.jar
-rw-rw-rw-. 1 1001 root  192762 Oct 25 02:26 mysql-binlog-connector-java-0.25.1.jar
-rw-rw-rw-. 1 1001 root 2397321 Oct 25 02:26 mysql-connector-java-8.0.21.jar
```

ポッドを再起動して、ローカルマシンからRESTでPluginを確認する。

```
>curl -H "Accept:application/json" https://connectors.test-ocp-f83035ec98caa9e1c443f354e8208d2f-0000.jp-tok.containers.appdomain.cloud:443/connector-plugins -k |jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   597  100   597    0     0    711      0 --:--:-- --:--:-- --:--:--   710
[
  {
    "class": "io.debezium.connector.mysql.MySqlConnector",
    "type": "source",
    "version": "1.6.0.Final"
  },
  {
    "class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "type": "sink",
    "version": "6.1.0-ce"
  },
  {
    "class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "type": "source",
    "version": "6.1.0-ce"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
    "type": "source",
    "version": "1"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
    "type": "source",
    "version": "1"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "type": "source",
    "version": "1"
  }
]
```

io.debezium.connector.mysql.MySqlConnectorが表示されている。
（他のPluginはデフォルトで入っていたもの）

Confluent Control Center(C3)からも見える。
![Connector_2021-10-26-11-06-24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/682553/ec983714-7770-c7da-f160-7ced305c46de.png)

## 2. MySQL Source Connectorの構成

C3からも構成できるが、ここではローカルマシンからRESTでConnectorを構成

```
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" -k https://connectors.test-ocp-f83035ec98caa9e1c443f354e8208d2f-0000.jp-tok.containers.appdomain.cloud:443/connectors/ -d '{ "name": "mysql-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "172.21.17.38", "database.port": "3306", "database.user": "user", "database.password": "pass", "database.server.id": "184055", "database.server.name": "dbserver1", "database.include.list": "opdb", "database.history.kafka.bootstrap.servers": "kafka:9071", "database.history.kafka.topic": "dbhistory.opdb", "table.include.list": "opdb.menus", "database.history.consumer.security.protocol": "SASL_SSL", "database.history.consumer.sasl.mechanism": "PLAIN", "database.history.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user\" password=\"password\";", "database.history.producer.security.protocol": "SASL_SSL", "database.history.producer.sasl.mechanism": "PLAIN", "database.history.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user\" password=\"password\";", "include.schema.changes": "true" } }'
```

データ部を整形して表示

```json
{
	"name": "mysql-connector",
	"config": {
		"connector.class": "io.debezium.connector.mysql.MySqlConnector",
		"tasks.max": "1",
		"database.hostname": "172.21.17.38",
		"database.port": "3306",
		"database.user": "user",
		"database.password": "pass",
		"database.server.id": "184055",
		"database.server.name": "dbserver1",
		"database.include.list": "opdb",
		"database.history.kafka.bootstrap.servers": "kafka:9071",
		"database.history.kafka.topic": "dbhistory.opdb",
		"table.include.list": "opdb.menus",
		"database.history.consumer.security.protocol": "SASL_SSL",
		"database.history.consumer.sasl.mechanism": "PLAIN",
		"database.history.consumer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user\" password=\"password\";",
		"database.history.producer.security.protocol": "SASL_SSL",
		"database.history.producer.sasl.mechanism": "PLAIN",
		"database.history.producer.sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user\" password=\"password\";",
		"include.schema.changes": "true"
	}
}
```

プロパティは以下を参照。

[MySQL connector configuration example](https://debezium.io/documentation/reference/1.6/connectors/mysql.html#mysql-example-configuration)
※Connectorのバージョンによってプロパティが変わることがあるので、使用しているバージョンのドキュメントを参照すること。ここでは1.6を使用。

基本のプロパティは上記リンクで確認できるが、Kafkaサーバーに対してSASL/PLAINで接続するためのプロパティ(database.history.producer.xxx、database.history.consumer.xxx)を追加している。

[Debezium MySQL Source Connector Configuration Properties - Database History Parameters](https://docs.confluent.io/debezium-connect-mysql-source/current/mysql_source_connector_config.html#database-history-parameters)

構成したSource Connectorが正常に稼働したことを確認。

```
>curl -H "Accept:application/json" https://connectors.test-ocp-f83035ec98caa9e1c443f354e8208d2f-0000.jp-tok.containers.appdomain.cloud:443/connectors/mysql-connector/status -k |jq
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   245  100   245    0     0    641      0 --:--:-- --:--:-- --:--:--   641
{
  "name": "mysql-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "connectors-1.connectors.confluent-namespace.svc:9083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "connectors-1.connectors.confluent-namespace.svc:9083"
    }
  ],
  "type": "source"
}
```

# トピック上のデータの確認

構成したConnectorによってMySQLのデータがKafkaのトピックに送信されたことを確認する。

まず、Kafka上に作成されたトピックを確認する。

以下の3つのトピックが作成されている。
![Connector_2021-12-12-12-58-36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/682553/15d634a7-2272-db67-8f3a-aa55d8e6af64.png)

odserver1.opdb.menusトピックに以下のデータが送信されている。

```json
[
	{
		"topic": "dbserver1.opdb.menus",
		"partition": 0,
		"offset": 0,
		"timestamp": 1639280629029,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"before": null,
			"after": {
				"id": 1,
				"name": "curry"
			},
			"source": {
				"version": "1.6.0.Final",
				"connector": "mysql",
				"name": "dbserver1",
				"ts_ms": 1639280628022,
				"snapshot": "true",
				"db": "opdb",
				"sequence": null,
				"table": "menus",
				"server_id": 0,
				"gtid": null,
				"file": "binlog.000002",
				"pos": 3264,
				"row": 0,
				"thread": null,
				"query": null
			},
			"op": "r",
			"ts_ms": 1639280628024,
			"transaction": null
		}
	},
	{
		"topic": "dbserver1.opdb.menus",
		"partition": 0,
		"offset": 1,
		"timestamp": 1639280629030,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"before": null,
			"after": {
				"id": 2,
				"name": "hamburger"
			},
			"source": {
				"version": "1.6.0.Final",
				"connector": "mysql",
				"name": "dbserver1",
				"ts_ms": 1639280628026,
				"snapshot": "true",
				"db": "opdb",
				"sequence": null,
				"table": "menus",
				"server_id": 0,
				"gtid": null,
				"file": "binlog.000002",
				"pos": 3264,
				"row": 0,
				"thread": null,
				"query": null
			},
			"op": "r",
			"ts_ms": 1639280628026,
			"transaction": null
		}
	}
]
```

Offset=0にmenusテーブルの1行目のデータ、Offset=1に2行目のデータが入っていることが分かる。

menusテーブルに行を追加すると、トピックにも新しいデータが送信される。

```sql
INSERT INTO menus (id, name) VALUES (3, 'katsudon');
```

```json
	{
		"topic": "dbserver1.opdb.menus",
		"partition": 0,
		"offset": 2,
		"timestamp": 1639280629030,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"before": null,
			"after": {
				"id": 3,
				"name": "katsudon"
			},
			"source": {
				"version": "1.6.0.Final",
				"connector": "mysql",
				"name": "dbserver1",
				"ts_ms": 1639280628026,
				"snapshot": "true",
				"db": "opdb",
				"sequence": null,
				"table": "menus",
				"server_id": 0,
				"gtid": null,
				"file": "binlog.000002",
				"pos": 3264,
				"row": 0,
				"thread": null,
				"query": null
			},
			"op": "r",
			"ts_ms": 1639280628026,
			"transaction": null
		}
	}
```

テーブル内のデータをUPDATEした場合は、以下のメッセージがトピックに送信される。

```sql
UPDATE menus SET name='tendon' WHERE id=3;
```

```json
[
	{
		"topic": "dbserver1.opdb.menus",
		"partition": 0,
		"offset": 4,
		"timestamp": 1639530256773,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"before": {
				"id": 3,
				"name": "katsudon"
			},
			"after": {
				"id": 3,
				"name": "tendon"
			},
			"source": {
				"version": "1.6.0.Final",
				"connector": "mysql",
				"name": "dbserver1",
				"ts_ms": 1639530256000,
				"snapshot": "false",
				"db": "opdb",
				"sequence": null,
				"table": "menus",
				"server_id": 1,
				"gtid": null,
				"file": "binlog.000002",
				"pos": 3486,
				"row": 0,
				"thread": null,
				"query": null
			},
			"op": "u",
			"ts_ms": 1639530256515,
			"transaction": null
		}
	}
]
```

beforeにはUPDATE前のデータ、afterにはUPDATE後のデータが入っている。

DELETEの場合は以下。

```sql
DELETE FROM menus WHERE id=4;
```

```json
[
	{
		"topic": "dbserver1.opdb.menus",
		"partition": 0,
		"offset": 5,
		"timestamp": 1639530687264,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"before": {
				"id": 4,
				"name": "gyoza"
			},
			"after": null,
			"source": {
				"version": "1.6.0.Final",
				"connector": "mysql",
				"name": "dbserver1",
				"ts_ms": 1639530687000,
				"snapshot": "false",
				"db": "opdb",
				"sequence": null,
				"table": "menus",
				"server_id": 1,
				"gtid": null,
				"file": "binlog.000002",
				"pos": 3794,
				"row": 0,
				"thread": null,
				"query": null
			},
			"op": "d",
			"ts_ms": 1639530687221,
			"transaction": null
		}
	}
]
```

Connectorのdatabase.history.kafka.topicプロパティで指定したdbhistory.opdbトピックには以下のメッセージが格納されている。
このトピックは、Connectorがデータベースのスキーマ履歴を保管するために利用される。

```json
[
	{
		"topic": "dbhistory.opdb",
		"partition": 0,
		"offset": 0,
		"timestamp": 1639280627942,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"source": {
				"server": "dbserver1"
			},
			"position": {
				"ts_sec": 1639280627,
				"file": "binlog.000002",
				"pos": 3264,
				"snapshot": true
			},
			"databaseName": "",
			"ddl": "SET character_set_server=utf8mb4, collation_server=utf8mb4_0900_ai_ci",
			"tableChanges": []
		}
	},
	{
		"topic": "dbhistory.opdb",
		"partition": 0,
		"offset": 1,
		"timestamp": 1639280627970,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"source": {
				"server": "dbserver1"
			},
			"position": {
				"ts_sec": 1639280627,
				"file": "binlog.000002",
				"pos": 3264,
				"snapshot": true
			},
			"databaseName": "opdb",
			"ddl": "DROP TABLE IF EXISTS `opdb`.`menus`",
			"tableChanges": []
		}
	},
	{
		"topic": "dbhistory.opdb",
		"partition": 0,
		"offset": 2,
		"timestamp": 1639280627973,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"source": {
				"server": "dbserver1"
			},
			"position": {
				"ts_sec": 1639280627,
				"file": "binlog.000002",
				"pos": 3264,
				"snapshot": true
			},
			"databaseName": "opdb",
			"ddl": "DROP DATABASE IF EXISTS `opdb`",
			"tableChanges": []
		}
	},
	{
		"topic": "dbhistory.opdb",
		"partition": 0,
		"offset": 3,
		"timestamp": 1639280627975,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"source": {
				"server": "dbserver1"
			},
			"position": {
				"ts_sec": 1639280627,
				"file": "binlog.000002",
				"pos": 3264,
				"snapshot": true
			},
			"databaseName": "opdb",
			"ddl": "CREATE DATABASE `opdb` CHARSET utf8mb4 COLLATE utf8mb4_0900_ai_ci",
			"tableChanges": []
		}
	},
	{
		"topic": "dbhistory.opdb",
		"partition": 0,
		"offset": 4,
		"timestamp": 1639280627981,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"source": {
				"server": "dbserver1"
			},
			"position": {
				"ts_sec": 1639280627,
				"file": "binlog.000002",
				"pos": 3264,
				"snapshot": true
			},
			"databaseName": "opdb",
			"ddl": "USE `opdb`",
			"tableChanges": []
		}
	},
	{
		"topic": "dbhistory.opdb",
		"partition": 0,
		"offset": 5,
		"timestamp": 1639280627998,
		"timestampType": "CREATE_TIME",
		"headers": [],
		"key": null,
		"value": {
			"source": {
				"server": "dbserver1"
			},
			"position": {
				"ts_sec": 1639280627,
				"file": "binlog.000002",
				"pos": 3264,
				"snapshot": true
			},
			"databaseName": "opdb",
			"ddl": "CREATE TABLE `menus` (\n  `id` int DEFAULT NULL,\n  `name` varchar(100) DEFAULT NULL\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci",
			"tableChanges": [
				{
					"type": "CREATE",
					"id": "\"opdb\".\"menus\"",
					"table": {
						"defaultCharsetName": "utf8mb4",
						"primaryKeyColumnNames": [],
						"columns": [
							{
								"name": "id",
								"jdbcType": 4,
								"typeName": "INT",
								"typeExpression": "INT",
								"charsetName": null,
								"position": 1,
								"optional": true,
								"autoIncremented": false,
								"generated": false
							},
							{
								"name": "name",
								"jdbcType": 12,
								"typeName": "VARCHAR",
								"typeExpression": "VARCHAR",
								"charsetName": "utf8mb4",
								"length": 100,
								"position": 2,
								"optional": true,
								"autoIncremented": false,
								"generated": false
							}
						]
					}
				}
			]
		}
	}
]
```


# おわりに

本来はSink Connectorも構成して、他のデータベースにレプリケーションするつもりでしたが、OCP環境が諸般の事情で使えなくなったので、OCP環境が復活したら、再開しようと思います。