メッセージフィルタリングを導入する

[kohei-kohei]

概要

メッセージフィルタリングを実装する

参考

https://debezium.io/documentation/reference/2.4/transformations/filtering.html

動作確認

データの登録・更新・削除を行う

docker exec mysql mysql -u root -ppassword playground -e "INSERT INTO items (name) VALUE ('Tシャツ');"
docker exec mysql mysql -u root -ppassword playground -e "UPDATE items SET name = 'ロングスリーブTシャツ' WHERE id = 1;"
docker exec mysql mysql -u root -ppassword playground -e "DELETE FROM items WHERE id = 1;"

トピックの中を確認すると登録と更新のメッセージのみが入っていることが分かる

docker exec kafka-broker kafka-console-consumer --bootstrap-server kafka-broker:9092 \
       --topic debezium_playground_topic.playground.items --from-beginning

{"before":null,"after":{"id":1,"name":"Tシャツ","registered_at":1702272672505,"updated_at":1702272672505},"source":{"version":"2.4.1.Final","connector":"mysql","name":"debezium_playground_topic","ts_ms":1702272672000,"snapshot":"false","db":"playground","sequence":null,"table":"items","server_id":1,"gtid":null,"file":"mysql-bin.000003","pos":398,"row":0,"thread":12,"query":null},"op":"c","ts_ms":1702272672528,"transaction":null}
{"before":{"id":1,"name":"Tシャツ","registered_at":1702272672505,"updated_at":1702272672505},"after":{"id":1,"name":"ロングスリーブTシャツ","registered_at":1702272672505,"updated_at":1702272690558},"source":{"version":"2.4.1.Final","connector":"mysql","name":"debezium_playground_topic","ts_ms":1702272690000,"snapshot":"false","db":"playground","sequence":null,"table":"items","server_id":1,"gtid":null,"file":"mysql-bin.000003","pos":748,"row":0,"thread":14,"query":null},"op":"u","ts_ms":1702272690571,"transaction":null}