Kafka Connect の概念 ¶

デッドレターキュー ¶

さまざまな理由から、無効なレコードが生じることがあります。1 例として、シンクコネクターの構成で Avro フォーマットが想定されているにもかかわらず、そのシンクコネクターに JSON フォーマットでシリアル化されたレコードが届いた場合が挙げられます。無効なレコードをシンクコネクターで処理できない場合、コネクターの構成プロパティ errors.tolerance に基づいてエラーが処理されます。

デッドレターキューは、シンクコネクターにのみ適用されます。

デッドレターキューのトピックは、Confluent Cloud シンクコネクター用に自動生成されます。詳細については、「 Confluent Cloud デッドレターキュー 」を参照してください。

この構成プロパティの有効な値としては、 none （デフォルト）と all の 2 つがあります。

errors.tolerance が none に設定されている場合、エラーが発生するか無効なレコードが見つかると、その時点でコネクタータスクは失敗となり、コネクターのステータスが Failed になります。この問題を解決するには、Kafka Connect ワーカーのログを確認して障害が発生した原因を特定し、問題を修正して、コネクターを再起動する必要があります。

errors.tolerance が all に設定されている場合、エラーや無効なレコードはすべて無視され、処理が続行されます。Connect ワーカーのログには、エラーは書き込まれません。失敗したレコードがあるかどうかを判断するには、 内部メトリクス を使用するか、ソースのレコード数を計算して、処理されたレコード数と比較する必要があります。

エラー処理機能を使用することができ、その場合、無効なレコードはすべて特別なトピックにルーティングされ、エラーがレポートされます。このトピックには、シンクコネクターで処理できなかったレコードの デッドレターキュー が含まれます。

デッドレターキューのトピックの作成 ¶

デッドレターキューを作成するには、以下の構成プロパティをシンクコネクターの構成に追加します。

errors.tolerance = all
errors.deadletterqueue.topic.name = <dead-letter-topic-name>
デッドレターキューが有効になっている、GCS のシンクコネクターの構成の例を以下に示します。
  "name": "gcs-sink-01",
  "config": {
    "connector.class": "io.confluent.connect.gcs.GcsSinkConnector",
    "tasks.max": "1",
    "topics": "gcs_topic",
    "gcs.bucket.name": "<my-gcs-bucket>",
    "gcs.part.size": "5242880",
    "flush.size": "3",
    "storage.class": "io.confluent.connect.gcs.storage.GcsStorage",
    "format.class": "io.confluent.connect.gcs.format.avro.AvroFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081",
    "schema.compatibility": "NONE",
    "confluent.topic.bootstrap.servers": "localhost:9092",
    "confluent.topic.replication.factor": "1",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq-gcs-sink-01"
デッドレタートピックには失敗したレコードが含まれていますが、それでは、理由はわかりません。さらに次の構成プロパティを追加すると、失敗したレコードのヘッダー情報を含めることができます。
errors.deadletterqueue.context.headers.enable = true
Record headers are added to the dead letter queue when this parameter is set to true
(the default is false). You can then use the kcat （旧 kafkacat）ユーティリティ to view the record header
and determine why the record failed. Errors are also sent to Connect Reporter.
元のレコードヘッダーとの競合を回避するため、デッドレターキューのコンテキストヘッダーキーは先頭が _connect.errors になります。
先ほどの例で、ヘッダーを有効にした構成がこちらです。
  "name": "gcs-sink-01",
  "config": {
    "connector.class": "io.confluent.connect.gcs.GcsSinkConnector",
    "tasks.max": "1",
    "topics": "gcs_topic",
    "gcs.bucket.name": "<my-gcs-bucket>",
    "gcs.part.size": "5242880",
    "flush.size": "3",
    "storage.class": "io.confluent.connect.gcs.storage.GcsStorage",
    "format.class": "io.confluent.connect.gcs.format.avro.AvroFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081",
    "schema.compatibility": "NONE",
    "confluent.topic.bootstrap.servers": "localhost:9092",
    "confluent.topic.replication.factor": "1",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq-gcs-sink-01",
    "errors.deadletterqueue.context.headers.enable":true
このトピックの詳細については、「Kafka Connect の詳説 – エラー処理とデッドレターキュー」を参照してください。
セキュリティを有効にしたデッドレターキューの使用¶
セキュリティを有効にして Confluent Platform を使用すると、Confluent Platform 管理クライアント によってデッドレターキューのトピックが作成されます。無効なレコードは、まず、それらのレコードを送信するように構成されている内部のプロデューサーに渡されます。その後、管理クライアントによってデッドレターキューのトピックが作成されます。
セキュアな Confluent Platform 環境でデッドレターキューを使用するには、追加の管理クライアントの構成プロパティ（.admin というプレフィックス付き）を Connect ワーカーの構成に追加する必要があります。追加の Connect ワーカーの構成プロパティを指定する SASL/PLAIN の例を以下に示します。
admin.ssl.endpoint.identification.algorithm=https
admin.sasl.mechanism=PLAIN
admin.security.protocol=SASL_SSL
admin.request.timeout.ms=20000
admin.retry.backoff.ms=500
admin.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="<user>" \
  password="<secret>";
ロールベースアクセス制御（RBAC）環境における Connect ワーカー、シンクコネクター、デッドレターキュートピックの構成の詳細については、「Kafka Connect と RBAC」を参照してください。
おすすめの関連情報¶
ブログ記事: Kafka Connect の詳説 – エラー処理とデッドレターキュー
    
Confluent Cloud is a fully-managed Apache Kafka service available on all three major clouds.
      Try it free today.
    Get Started Free