TiDB Essential : Changefeed 機能を試す Kafka 版
過去こちらの記事でTiDB Essentail が新たにサポートしたChange Feed機能を試しました。
記事の中ではChange Feedの送出先を Amazon RDS/MySQLに指定しました。
ChangeFeed機能はこのほかにApache Kafkaに対応しているのでその設定手順を以下に纏めます。
Apache Kafka とは
Apache Kafka（アパッチ カフカ）は、LinkedInによって開発され、現在はApache Software Foundationによって管理されているオープンソースの分散ストリーミングプラットフォームです。Kafkaは、リアルタイムのデータパイプラインやストリーム処理アプリケーションを構築するために広く使われており、膨大なデータの取り込み・保存・処理・転送を高いスループットで実現します。
その特徴は、「パブリッシュ/サブスクライブ型メッセージング」「分散設計」「高いスケーラビリティ」「永続性」 にあり、ログ収集、監査データの蓄積、IoTイベントの処理、ユーザー行動のリアルタイム解析、マイクロサービス間の非同期通信など、さまざまな用途に活用されています。
Kafkaは「メッセージキュー」としての役割を超え、データストリーミングの中核基盤として、今日のデータ駆動型アーキテクチャに不可欠な存在となっています。
AWS では Kafka のマネージドサービスである、 Amazon Managed Streaming for Apache Kafka を提供していますが、今日はせっかくなので Kafka を EC2 にインストールすることで小規模な環境で中身を見ていきたいと思います。
さっそくやつてみる
1. Amazon EC2 の起動
この手順は割愛しますがパブリックサブネットで外部からアクセス可能なEC2を起動します。Kafkaはデフォルトで 9092 ポートを使用しますので、そのポートを開けておきます。
なおKafkaはデフォルトでは最低1GBのメモリを必要としますので、2GB以上のインスタンスを起動します。
2. Java と Kafka のインストール
まずはAmazon Correto (OpenJDK)をインストールします。
sudo dnf install java-17-amazon-corretto-devel -y
java -version
次にKafkaのダウンロードとインストールを行います。
cd /opt
sudo wget https://downloads.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz
sudo tar -xzf kafka_2.13-3.9.0.tgz
sudo mv kafka_2.13-3.9.0 kafka
sudo chown -R $USER:$USER /opt/kafka
3. Kafka の待ち受けホスト名の変更
Kafkaの設定はデフォルトでは localhost のみアクセス可能となっています。これをTiDB側からアクセスできるようにするために、EC2のホスト名でのアクセスを受けるけれるようにします。
cd ./config
vi server.properties
以下の2行を追記します。
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://ec2-57-183-26-185.ap-northeast-1.compute.amazonaws.com:9092
ec2-57-183-26-185は皆さんの環境ごとに変更してください。
4. Kafka の起動
cd ..
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
sleep 5
bin/kafka-server-start.sh -daemon config/server.properties
sleep 10
jps
以下の様に表示されていれば成功です。
26593 QuorumPeerMain
28406 Kafka
28487 Jps
5. テスト用メッセージの送信と受信
次に以下でテスト用チャネルの作成とメッセージの受信待機設定を実行します。
bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 5 --replication-factor 1
bin/kafka-console-consumer.sh --topic test-topic --from-beginning --bootstrap-server localhost:9092
次にもう一つターミナルを開きメッセージを送信します。
bin/kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092
入力プロンプトで適当な文字列を入力すると元のターミナルでは文字列を受信するはずです。
6. TiDB Essential で Change Feed の設定
左ペイン Data→ChangeFeedを想定します。
Create ChangeFeedをクリックします。
Kafkaを選択し、Kafkaが動作しているEC2のホスト名を入力します。
以下はデフォルトのまま Next をクリックします。
先ほど作成したテスト用トピックを指定します。
ChangeFeedを受け取るKafkaクラスターの情報（先ほどEC2に起動したKafka）に合わせて数字を修正します。この手順であれば両方1になります。
Next→Submitをクリックします。
ChangeFeedが作成中になりますので完了するまで待ちます。
7. テスト
では以下のSQLを実行してみます。
-- テーブル作成
CREATE TABLE users (
id INT PRIMARY KEY,
name VARCHAR(50),
email VARCHAR(100),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- データ挿入
INSERT INTO users (id, name, email) VALUES
(1, 'Taro Yamada', 'taro@example.com'),
(2, 'Hanako Suzuki', 'hanako@example.com'),
(3, 'Jiro Tanaka', 'jiro@example.com');
-- 確認
SELECT * FROM users;
+----+---------------+--------------------+---------------------+
| id | name | email | created_at |
+----+---------------+--------------------+---------------------+
| 1 | Taro Yamada | taro@example.com | 2026-02-05 06:22:50 |
| 2 | Hanako Suzuki | hanako@example.com | 2026-02-05 06:22:50 |
| 3 | Jiro Tanaka | jiro@example.com | 2026-02-05 06:22:50 |
+----+---------------+--------------------+---------------------+
3 rows in set (0.02 sec)
以下の様にKafka側でイベントを受信します。
{"id":0,"database":"test","table":"users","pkNames":["id"],"isDdl":false,"type":"INSERT","es":1770275748661,"ts":1770275750856,"sql":"","sqlType":{"id":4,"name":12,"email":12,"created_at":93},"mysqlType":{"email":"varchar","created_at":"timestamp","id":"int","name":"varchar"},"old":null,"data":[{"id":"1","name":"Taro Yamada","email":"taro@example.com","created_at":"2026-02-05 07:15:48"}]}
{"id":0,"database":"test","table":"users","pkNames":["id"],"isDdl":false,"type":"INSERT","es":1770275748661,"ts":1770275750856,"sql":"","sqlType":{"id":4,"name":12,"email":12,"created_at":93},"mysqlType":{"id":"int","name":"varchar","email":"varchar","created_at":"timestamp"},"old":null,"data":[{"id":"2","name":"Hanako Suzuki","email":"hanako@example.com","created_at":"2026-02-05 07:15:48"}]}
{"id":0,"database":"test","table":"users","pkNames":["id"],"isDdl":false,"type":"INSERT","es":1770275748661,"ts":1770275750856,"sql":"","sqlType":{"id":4,"name":12,"email":12,"created_at":93},"mysqlType":{"id":"int","name":"varchar","email":"varchar","created_at":"timestamp"},"old":null,"data":[{"id":"3","name":"Jiro Tanaka","email":"jiro@example.com","created_at":"2026-02-05 07:15:48"}]}
CREATE TABLEは同期されませんので注意が必要です。
Discussion