論理デコード出力プラグイン †
要約 †
Bottled WaterのKafkaログ圧縮とAvroデータエンコーディングの使用を検討したところで、Postgresとの統合の内部を少し見て、それがどのように論理デコード機能を使用しているかを見てみよう。
図 3-9. どうやってソーセージが作られるのか、もしくは水がボトルに入れられるのか
Postgresの論理デコード機能の興味深い特性は、変更データがネットワークを介してコンシューマに送られるワイヤフォーマットを定義していないことである。代わりに挿入や更新、削除された行ごとに関数呼び出しを受け取る出力プラグインAPIを定義する。 Bottled WaterはこのAPIを使用してデータベースの内部形式のデータを読み取り、Avroにシリアライズする。
出力プラグインは、Postgres拡張機構を使用してC言語で記述し、共有ライブラリとしてデータベースサーバにロードする必要がある（図 3-10）。これには、スーパーユーザー権限とデータベースサーバー上のファイルシステムアクセスが必要なため、軽く引き受けるべきことではない。私は、多くのデータベース管理者がデータベースサーバー内でカスタムコードを実行する可能性があることを怖がっていることを理解している。残念ながら、これは論理デコードが現在使用できる唯一の方法である。
現時点では、リーダーデータベースに論理デコードプラグインをインストールする必要がある。原則として、それは他のクライアントに影響を与えないように別個のフォロワー上で実行させることが可能ですが、Postgresの現在の実装はこれを許可していない。この制約はPostgresの将来のバージョンで対応されるだろう。
図 3-10. Bottled Waterプラグインは、データベースサーバー内で実行される。クライアントデーモンはDBに接続し、スキーマをレジストリに送信し、データをKafkaに送信する。
クライアントデーモン †
Bottled Waterはデータベースサーバー内で実行されるプラグインのほかに、どこでも実行できるクライアントプログラムで構成されている。これはPostgresサーバーとKafkaブローカに接続し、データベースからAvroエンコードされたデータを受信し、それをKafkaに転送する。
クライアントはC言語でも書かれており、これはPostgresクライアントライブラリをそのように使用するのが最も簡単で、プラグインとクライアントの間でいくつかのコードが共有されるためである。かなり軽量で、ディスクに書き込む必要はない。執筆時点では、Bottled WaterクライアントとKafka Connectフレームワークを統合する作業が進められている。
クライアントがクラッシュしたり、PostgresまたはKafkaから切断された場合はどうなるのだろうか？ 問題ない。Kafkaのブローカーによってどのメッセージが発行され、確認されたかを追跡する。クライアントはエラーの後で再起動すると、確認応答されていないすべてのメッセージを再実行する。したがって、Kafkaではいくつかのメッセージが2回出現する可能性があるが、データは失われない。ログの圧縮により、最終的に重複したメッセージが削除される。
並行性 †
もう1つ疑問が残っている。複数のクライアントが同時にデータベースに書き込むとどうなるのか？(図3-11) これらの書き込みの結果、Kafkaに送られる変更ストリームにどのように反映されるのか？トランザクションがデータを書き込んだ後、コミットする前に中止するとどうなるのか？
図 3-11. 2つのトランザクションが同時にデータベースに書き込むが、Bottled Waterはコミットされた順番に変更がコミットされたことしか見えない。
幸いBottled Waterの場合、PostgreSQLの論理デコードAPIは簡単な答えを提供している。トランザクション中に行われたすべての書き込みは、トランザクションがコミットするときに同時に論理デコードAPIに公開される。これは、Bottled Waterが中止されたトランザクション（後で中止されるトランザクションによって作成された書き込みも表示されない）や書き込みの順序付けについて心配する必要がないことを意味している。
PostgreSQLのトランザクション分離セマンティクスは、コミットされた順序で書き込みを適用すると、正しい結果を得ることができる。しかし、WALは実際にはいくつかの異なるトランザクションからのインターリーブされた書き込みを含むことがある。したがって、WALをデコードする際、論理デコード機能は、トランザクションコミットの順番で表示されるように、それらの書き込みを並べ替える必要がある。
Postgresは、変更データの取り込みのこの特定の側面を容易にする。変更データの取り込みを別のデータベースで実装する場合は、これらの並行性の問題を自分で処理する必要がある。
担当者のつぶやき †
みんなの突っ込み †
