6.5. Debezium コネクターへのシグナル送信

シグナルデータコレクションまたはシグナルテーブルは、コネクターに送信して指定の操作をトリガーするシグナルを保存します。シグナリングテーブルの構造は、以下の標準フォーマットに準拠する必要があります。

標準 SQL の DDL クエリーをソースデータベースに送信して、シグナリングテーブルを作成します。

次の例は、3 列の debezium_signal テーブルを作成する CREATE TABLE コマンドです。

CREATE TABLE debezium_signal (id VARCHAR(42) PRIMARY KEY, type VARCHAR(32) NOT NULL, data VARCHAR(2048) NULL);

log シグナルタイプのシグナリングテーブルエントリーを作成することで、ログにエントリーを追加するようコネクターに要求できます。シグナルの処理後、コネクターは指定されたメッセージをログに出力します。オプションとして、結果として得られるメッセージにストリーミング座標が含まれるようにシグナルを設定することもできます。

{"message": "Signal message at offset {}"}

execute-snapshot シグナルタイプのシグナルを作成すると、アドホックスナップショットの開始をコネクターに要求できます。信号を処理した後、コネクターは要求されたスナップショットオペレーションを実行します。

コネクターが最初に起動したときに実行される最初のスナップショットとは異なり、アドホックスナップショットは、コネクターがすでにデータベースからの変更イベントのストリーミングを開始した後のランタイム中に発生します。いつでもアドホックなスナップショットを開始することができます。

アドホックスナップショットは、以下の Debezium コネクターで利用可能です。

{"data-collections": ["public.MyFirstTable", "public.MySecondTable"]}

{"type":"execute-snapshot","data": {"data-collections": ["public.MyFirstTable"], "type": "INCREMENTAL", "additional-conditions":[{"data-collection": "public.MyFirstTable", "filter":"color='blue' AND brand='MyBrand'"}]}}

アドホックスナップショットの詳細は、お使いのコネクターのドキュメントの スナップショット のトピックを参照してください。

次の Debezium コネクターのアドホックスナップショットを停止できます。

{"type":"INCREMENTAL", "data-collections": ["public.MyFirstTable"]}

シグナルの type を指定する必要があります。data-collections フィールドはオプションです。現在のスナップショットのすべてのアクティビティーを停止するようコネクターに要求するには、data-collections フィールドを空白のままにします。増分スナップショットを続行したいが、スナップショットから特定のコレクションを除外したい場合は、除外するコレクションまたは正規表現の名前のコンマ区切りリストを指定します。コネクターが信号を処理した後、増分スナップショットが続行されますが、指定したコレクションからデータが除外されます。

増分スナップショットはアドホックスナップショットの一種です。増分スナップショットでは、初期スナップショットと同様に、指定したテーブルのベースライン状態をキャプチャします。しかし、初期スナップショットとは異なり、増分スナップショットでは、一度にすべてのテーブルをキャプチャーするのではなく、チャンク単位でテーブルをキャプチャします。このコネクターは、スナップショットの進捗状況を追跡するために、電子透かしを使用しています。

増分スナップショットは、指定されたテーブルの初期状態を単一のモノリシックな操作ではなくチャンクでキャプチャーすることにより、初期スナップショットのプロセスに比べて以下のような利点があります。

次の Debezium コネクターの増分スナップショットを一時停止できます。

シグナルの type を指定する必要があります。data フィールドは無視されます。

次の Debezium コネクターの増分スナップショットを再開できます。

シグナルの type を指定する必要があります。data フィールドは無視されます。

増分スナップショットの詳細は、お使いのコネクターのドキュメントのスナップショットのトピックを参照してください。

execute-snapshot シグナルタイプと、値 blocking を持つ data.type でシグナルを作成すると、アドホックブロッキングスナップショットの開始をコネクターに要求できます。信号を処理した後、コネクターは要求されたスナップショットオペレーションを実行します。

コネクターが最初に起動したときに実行される初期スナップショットとは異なり、アドホックブロッキングスナップショットは、コネクターがデータベースからの変更イベントのストリーミングを停止した後のランタイム中に発生します。アドホックブロッキングスナップショットはいつでも開始できます。

ブロッキングスナップショットは、以下の Debezium コネクターで利用可能です。

  {"type": "blocking", "data-collections": ["schema1.table1", "schema1.table2"], "additional-conditions": [{"data-collection": "schema1.table1", "filter": "SELECT * FROM [schema1].[table1] WHERE column1 = 0 ORDER BY column2 DESC"}, {"data-collection": "schema1.table2", "filter": "SELECT * FROM [schema1].[table2] WHERE column2 > 0"}]}

{"type":"execute-snapshot","data": {"type": "blocking"}

ブロッキングスナップショットの詳細は、お使いのコネクターのドキュメントの スナップショット のトピックを参照してください。

カスタムアクションを使用すると、Debezium シグナリングフレームワークを拡張して、デフォルトの実装では使用できないアクションをトリガーできます。カスタムアクションは複数のコネクターで使用できます。

カスタムシグナルアクションを定義するには、次のインターフェイスを定義する必要があります。

@FunctionalInterface
public interface SignalAction<P extends Partition> {

    /**
     * @param signalPayload the content of the signal
     * @return true if the signal was processed
     */
    boolean arrived(SignalPayload<P> signalPayload) throws InterruptedException;
}

io.debezium.pipeline.signal.actions.SignalAction は、シグナリングチャネルを介して送信されるメッセージペイロードを表す 1 つのパラメーターを持つ、単一のメソッドを公開します。

カスタムシグナリングアクションを定義した後、SPI インターフェイス io.debezium.pipeline.signal.actions.SignalActionProvider を使用して、カスタムアクションをシグナリングメカニズムで使用できるようにします。

public interface SignalActionProvider {

    /**
     * Create a map of signal action where the key is the name of the action.
     *
     * @param dispatcher the event dispatcher instance
     * @param connectorConfig the connector config
     * @return a concrete action
     */

    <P extends Partition> Map<String, SignalAction<P>> createActions(EventDispatcher<P, ? extends DataCollectionId> dispatcher, CommonConnectorConfig connectorConfig);
}

実装では、シグナルアクションのマップを返すはずです。マップキーはアクションの名前に設定します。キーはシグナルの type として使用されます。

カスタムアクション Java プロジェクトには、Debezium コアモジュールに対するコンパイル依存関係があります。プロジェクトの pom.xml ファイルに次のコンパイル依存関係を含めます。

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-core</artifactId>
    <version>${version.debezium}</version> 1
</dependency>

META-INF/services/io.debezium.pipeline.signal.actions.SignalActionProvider ファイルでプロバイダーの実装を宣言します。