EKS上のDebeziumでAuroraのフェイルオーバー時に自動復旧を実現する方法

こんにちは。プラットフォーム開発部でリードエンジニアをしている chance です。

今回は EKS で実行している Debezium で、Aurora の FailOver 時に自動復旧させる設定についてお話しします。また補足として、SpringBoot などの Java アプリケーションでの FailOver 対策についても触れています。

Debezium の FailOver 対策としては、EKS の livenessProbe（死活判定プローブ）で tasks の state が FAILED になっていないかどうかを監視する。

通常の Java アプリケーションでは DNS キャッシュ TTL を調整の上、コネクションプールのテストクエリで read-only（読み取り専用）を見分ける。

弊社ではイベント駆動のメッセージングのブローカーとして Kafka(AWS MSK) を利用しています。また、Kafka への書き込みとしては Outbox パターンを採用しており、Aurora から Kafka への Producer として Debezium を利用しています。

Outbox パターン

Outbox パターンによってアプリケーションからの業務データとメッセージングのトランザクションの一貫性が保証されます。

この記事をお読みいただく方に Debezium 自体のご説明は不要かもしれませんので、サラッと書くと、Debezium は CDC(Change Data Capture：変更データキャプチャ) のための OSS です。具体的には、データベースのトランザクションログ（たとえば PostgreSQL の WAL ログ）を読んで更新差分を検知し、Kafka にメッセージを投げ込んでくれます。CDC 自体はデータベースのレプリケーションなどに用いられるものですが、Outbox パターンの実装方法としても最適です。

Debezium は常にデータベースに接続しており FailOver 発生時は自動で接続復旧を試みます。しかし、無事に接続が復旧できた後にエラーになることがあります。今回はこの対処についてお話ししていきます。

Debezium は EKS 上の Pod で動かしており、データベースには Aurora PostgreSQL を利用しています。

CDC(Change Data Capture：変更データキャプチャ) 概要図

以下のバージョンを利用しています。

（話に関係ないであろうもののバージョンは省略）

Aurora は FailOver 時にも素早く復旧してくれます。とはいえ、クライアントからすると一度接続が切れることには変わりありません。Debezium には接続が切れた際にリトライする機構が備わっており、再接続できた場合は処理を再開します。（v1.9 あたりでリトライ周りの実装をいろいろと改善して頂いたみたいです。）

Debezium 1.9 のリリースノート

しかし、ここで問題が発生しました。接続が復旧して Debezium が処理を再開する際、前回までの読み込み位置を確認して Replication Slot を更新しようとするのですが、これに失敗して処理を停止してしまいます。必ず発生するわけではなく、開発環境で試していると何回かに 1 回は発生しました。

エラーが発生する際のログの抜粋は以下です。

2024-01-26 07:12:59,697 ERROR WorkerSourceTask{id=<タスク ID>} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted [org.apache.kafka.connect.runtime.WorkerTask] org.apache.kafka.connect.errors.ConnectException: Unable to update filtered publication <パブリケーション名> for <スキーマ名>.<テーブル名> at ...（省略） Caused by: org.postgresql.util.PSQLException: ERROR: cannot execute ALTER PUBLICATION in a read-only transaction at ...（省略） 2024-01-26 07:12:59,699 INFO || Stopping down connector [io.debezium.connector.common.BaseSourceTask]

PSQLException: ERROR: cannot execute ALTER PUBLICATION in a read-only transaction

さらに困ったことに、Debezium 自体（Java プロセス）が停止してくれれば新しい Pod が立ち上がってくれるのでまだよいのですが、Java プロセスは生きたままなので CDC のみが停止した状態になってしまい、リカバリ作業が必要になってしまいます。

Aurora は接続用のエンドポイントとしてライター用とリーダー用を用意してくれます。ライター用エンドポイントはライターのインスタンスに、リーダー用エンドポイントはリードレプリカのインスタンスに繋がります。

{クラスタ名}.cluster-xxxxx.{region}.rds.amazonaws.com

{クラスタ名}.cluster-ro-xxxxx.{region}.rds.amazonaws.com

当然、Aurora の挙動としては FailOver 後にエンドポイントの繋ぎ替えまでしてくれます。では、Debezium はなぜ FailOver 後にリーダーインスタンスに接続してしまうのでしょうか。

ひとつには、Java の DNS キャッシュの影響が考えられます。Aurora のエンドポイントのドメインの名前解決をすると最終的にはインスタンスの IP アドレスが返ります。このため、Java は実際にはインスタンスと TCP 接続を結んでいることになります。

Debezium から Aurora への接続

Java は DNS の結果を（デフォルトでは）無制限にキャッシュするため、接続のたびに名前解決をすることなく FailOver 後も FailOver 前と同じインスタンスに繋ぎにいったと推測されます。

Debezium から Aurora への接続（FailOver 後）

Java の DNS キャッシュ対策として、TTL（Time To Live：生存時間）の値を設定して定期的に DNS 問い合わせをやり直すというプラクティスがあります。

// SpringBoot アプリケーションで、プロパティで指定する場合 package xxx.yyy.zzz; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import java.security.Security; @SpringBootApplication public class DemoApplication { // Java は DNS キャッシュをデフォルトで永続保持してしまい Aurora の failover などに対応できないため、ttl を指定する。 static { Security.setProperty("networkaddress.cache.ttl", "10"); Security.setProperty("networkaddress.cache.negative.ttl", "10"); } public static void main(String[] args) { SpringApplication.run(DemoApplication.class, args); } }

また、コネクションプール（接続プーリング）のテストクエリを工夫する方法も有効なようです。こちらのブログを参考にさせていただきました。

tech.asoview.co.jp

HikariCP の場合 spring: datasource: hikari: # transaction_read_only=off なら成功、それ以外は失敗するクエリにしておく。1/0 とすると SQL 自体が成立しないので random() を利用している。 connection-test-query: select case when current_setting('transaction_read_only') = 'off' then 1 else random()/0 end

Consumer や WebAPI は SpringBoot で開発していたため、これらの設定をすることで FailOver 後も接続を失うことなく動作しました。

しかし、今回の Debezium の場合、TTL のオプションを指定しても事象解決しませんでした。また、コネクションプールに関する設定項目は見当たりませんでした。

JDBC 接続の初期化時に実行するクエリの設定は存在しますが、ここに上記のテストクエリを設定しても接続を再試行してくれず、うまく復旧できませんでした。

さらにソースコードを追いかけていたところ、こればかりに時間をかけすぎられないと判断し、一旦手を止めました。

視点を変え、異常状態を検知して Pod を再起動できないかという検討に切り替えます。

最初に思いついたのは、エラーログ監視をトリガーとして再起動する方法です。エラー内容は既明なのでこれを監視しておき、Pod の再起動を呼び出せないかと考えました。しかし、ログフィルターから Pod 再起動コマンドを発行するまでにかなり遠回りとなるため、構築・保守の負担が増大することと、EKS 外部からの操作のための権限設定も必要になりそうであることから、もう少しシンプルに解決したいところです。

ところで、Debezium は実体としては Kafka Connect の Connector として動作しています。Connector にヘルスチェック機能はないのだろうかということで確認してみると、

docs.confluent.io

GET /connectors/(string:name)/tasks/(int:taskid)/status

Connector と Task がそれぞれ state を持っているようです。

Debezium(Kafka Conncect)のstatus(平常時)

実際に FailOver を実行して状態遷移を見てみると、FailOver 直後に JDBC 接続を失い接続リトライ中は state は RUNNING です。

Debezium(Kafka Conncect)のstatus(FailOver時)

FailOver 完了後に例のごとくリードレプリカに更新を試みて例外が発生したところで、state は FAILED です。

Debezium(Kafka Conncect)のstatus(エラー時)

これがトリガーとして使えそうです。Kubernetes には Pod の状態を識別するための Probe という仕様があります。

起動したことを識別するための条件

再起動が必要であるかを識別するための条件

(APIなどの場合に)serviceから接続できる状態かを識別するための条件

今回の事象に対処する前は、startup も liveness も 8083 ポートの /

変更前 livenessProbe: httpGet: path: / port: 8083 scheme: HTTP timeoutSeconds: 2 successThreshold: 1 failureThreshold: 1 periodSeconds: 60

今回の調査により、FailOver が起因となって Tasks の状態が FAILED となったことが判明しました。

変更後の livenessProbe は以下の通りです：exec: command: - "sh" - "-c" - "curl -s http://localhost:8083/connectors/<コネクタ名>/tasks/0/status | grep -v '"state":"FAILED"'" timeoutSeconds: 2 successThreshold: 1 failureThreshold: 1 periodSeconds: 60

これにより、状態が FAILED となったケースでも対応可能となりました。

EKS で Debezium を利用している環境において、Aurora の FailOver 時に自動復旧する事例を紹介しました。ステータスの見分け方についてはさらに精査が必要かもしれませんが、稼働状況を監視しつつ対応を進めていく方針です。

今回の件がどなたかのお役に立てれば幸いです。

みらい翻訳では、エンジニアを募集しています。

ご興味のある方は、ぜひ下記リンクよりご応募・お問い合わせをお待ちしております。

miraitranslate.com