エラー処理と中断からの再開のプロパティ

Overview

次の構成設定を使用して、MongoDB Kafka ソースコネクタがエラーを発生したときにどのように動作するかを指定し、中断された読み取りの再開に関連する設定を指定します。

設定

名前	説明
mongo.errors.tolerance	タイプ: string 説明: コネクタがエラーに遭遇したときにメッセージのプロセシングを続行するかどうか。コネクタがエラーに遭遇したときにメッセージのプロセシングを停止して問題を報告するようにしたい場合は、これを `"none"` に設定します。コネクタがメッセージのプロセシングを継続し、発生したエラーを無視するようにする場合は、これを `"all"` に設定します。重要: このプロパティは、errors.tolerance Connect フレームワークプロパティをオーバーライドします。デフォルト: `"none"` 許容値: `"none"` または `"all"`
mongo.errors.log.enable	Type: ブール値説明: コネクタがログファイルにエラーを報告するかどうか。コネクタが検出したすべてのエラーをログに記録するには、これを`true`に設定します。コネクタが許容しないエラーをログに記録するには、これを`false`に設定します。`errors.tolerance`または`mongo.errors.tolerance`設定を使用して、コネクタが許容するエラーを指定できます。重要: このプロパティは、errors.log.enable Connect Frameworkプロパティをオーバーライドします。デフォルト: `false` 許容値: `true`または`false`
mongo.errors.dealterqueue.token.name	Type: string Description: The name of topic to use as the dead letter queue. If you specify a value, the connector writes invalid messages to the dead letter queue topic as extended JSON strings. If you leave this setting blank, the connector does not write invalid messages to any topic. 重要: このプロパティを有効にするには、`errors.tolerance` または `mongo.errors.tolerance` の設定を `"all"` に設定する必要があります。デフォルト: `""` 受け入れ可能な値: 有効な Kafka トピック名
distinct. partition.name	型: string 説明: 使用するカスタムオフセットパーティション名。このオプションを使用すると、既存のオフセットに無効な再開トークンが含まれている場合に、新しい変更ストリームを開始するようにコネクタに指示できます。この設定を空白のままにすると、コネクタは接続の詳細に基づいてデフォルトのパーティション名を使用します。オフセットパーティションに名前を付ける戦略を表示するには、「保存されたオフセットのリセット」を参照してください。デフォルト: `""` 許容値: string。パーティションの名前付けの詳細については、Apache Kafka API ドキュメントの SourceRecord を参照してください。
heartbeat.Interval.ms	型: long 説明: コネクタがハートビートメッセージを送信するまでの待機時間（ミリ秒）。指定された間隔でソースレコードが公開されない場合、コネクタはハートビートメッセージを送信します。このメカニズムにより、名前空間のデータ量が少ない場合でもコネクタの再開可能性が向上します。ハートビートメッセージには、`postBatchResumeToken` データフィールドが含まれています。このフィールドの値には、コネクタが変更ストリームから最後に読み取った MongoDB サーバー oplog エントリが含まれます。ハートビートメッセージを無効にするには、これを `0` に設定します。学ぶには、無効なレジュームトークンページの防止を参照してください。デフォルト: `0` 許容値: 整数
heartbeat.token.name	タイプ: string 説明: コネクターがハートビートメッセージを公開するトピック名。この機能を有効にするには、`heartbeat.interval.ms` 設定で正の値を入力する必要があります。デフォルト: `__mongodb_heartbeats` 許容値: 有効な Kafka トピック名

単一メッセージ変換によるハートビート

Kafka Connect 配置でハートビートを有効にして単一メッセージ変換（SMT）を指定する場合は、SMT からハートビートメッセージを除外する必要があります。 SNT は Kafka Connect の機能であり、ストリーム処理アプリケーションを配置しなくても、ソースコネクタを通過するメッセージの変換を指定できます。

SMT からハートビートメッセージを除外するには、述語を作成して SNT に適用する必要があります。述語は、変換を適用する前にメッセージが条件ステートメントと一致するかどうかを確認できる S付与の機能です。

次の構成では、デフォルトのハートビートトピックに送信されたハートビートメッセージに一致するIsHeartbeat述語を定義します。

predicates=IsHeartbeat
predicates.IsHeartbeat.type=org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.IsHeartbeat.pattern=__mongodb_heartbeats

次の構成では、前述の述語を使用して、 ExtractField変換からハートビートメッセージを除外します。

transforms=Extract
transforms.Extract.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.Extract.field=<the field to extract from your Apache Kafka key>
transforms.Extract.predicate=IsHeartbeat
transforms.Extract.negate=true
# apply the default key schema as the extract transformation requires a struct object
output.format.key=schema

前述の変換からハートビートメッセージを除外しない場合、コネクターはハートビートメッセージを処理すると次のエラーを発生させます。

ERROR WorkerSourceTask{id=mongo-source-0} Task threw an uncaught and unrecoverable exception. Task is being killed ...
...
Only Struct objects supported for [field extraction], found: java.lang.String

SMD の詳細については、Confluent からのKafka Connect で単一メッセージ変換を使用する方法を参照してください。

述語の詳細については、 Confluent のフィルター（Apache Kafka）を参照してください。

ExtractField 変換の詳細については、Confluent からの ExtractField を参照してください。

デフォルトのキースキーマの詳細については、「デフォルトのスキーマ」ページを参照してください。

戻る

スタートアップ

すべてのプロパティ