AMQ Streams Kafka Bridge の使用

手順

1. Kafka Bridge インストールアーカイブを展開していない場合は、任意のディレクトリーに展開します。

2. 設定プロパティーをパラメーターとして使用して、Kafka Bridge スクリプトを実行します。

   以下に例を示します。

   ./bin/kafka_bridge_run.sh --config-file=<path>/application.properties

3. インストールが成功したことをログで確認します。

   HTTP-Kafka Bridge started and listening on port 8080
   HTTP-Kafka Bridge bootstrap servers localhost:9092

手順

1. Kafka Bridge を使用して、作成したトピックに 3 つのメッセージを生成します。

   curl -X POST \
     http://localhost:8080/topics/bridge-quickstart-topic \
     -H 'content-type: application/vnd.kafka.json.v2+json' \
     -d '{
       "records": [
           {
               "key": "my-key",
               "value": "sales-lead-0001"
           },
           {
               "value": "sales-lead-0002",
               "partition": 2
           },
           {
               "value": "sales-lead-0003"
           }
       ]
   }'

   • sales-lead-0001 は、キーのハッシュに基づいてパーティションに送信されます。
   • sales-lead-0002 は、パーティション 2 に直接送信されます。
   • sales-lead-0003 は、ラウンドロビン方式を使用して bridge-quickstart-topic トピックのパーティションに送信されます。

2. リクエストが正常に行われると、Kafka Bridge は offsets 配列を 200 コードと application/vnd.kafka.v2+json の content-type ヘッダーとともに返します。各メッセージで、offsets 配列は以下を記述します。

   • メッセージが送信されたパーティション。

   • パーティションの現在のメッセージオフセット。

     レスポンス例

     #...
     {
       "offsets":[
         {
           "partition":0,
           "offset":0
         },
         {
           "partition":2,
           "offset":0
         },
         {
           "partition":0,
           "offset":1
         }
       ]
     }

手順

1. bridge-quickstart-consumer-group という名前の新しいコンシューマーグループに Kafka Bridge コンシューマーを作成します。

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "name": "bridge-quickstart-consumer",
       "auto.offset.reset": "earliest",
       "format": "json",
       "enable.auto.commit": false,
       "fetch.min.bytes": 512,
       "consumer.request.timeout.ms": 30000
     }'

   • コンシューマーには bridge-quickstart-consumer という名前を付け、埋め込みデータ形式は json として設定します。
   • 一部の基本的な設定が定義されます。

   • コンシューマーはログへのオフセットに自動でコミットしません。これは、enable.auto.commit が false に設定されているからです。このクイックスタートでは、オフセットを後で手作業でコミットします。

     リクエストが正常に行われると、Kafka Bridge はレスポンス本文でコンシューマー ID (instance_id) とベース URL (base_uri) を 200 コードとともに返します。

     レスポンス例

     #...
     {
       "instance_id": "bridge-quickstart-consumer",
       "base_uri":"http://<bridge_id>-bridge-service:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer"
     }

2. ベース URL (base_uri) をコピーし、このクイックスタートの他のコンシューマー操作で使用します。

手順

1. トピックおよびパーティションへのメッセージの作成 の説明に従い、Kafka Bridge コンシューマーに新たなメッセージを生成します。

2. GET リクエストを records エンドポイントに送信します。

   curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records \
     -H 'accept: application/vnd.kafka.json.v2+json'

   Kafka Bridge コンシューマーを作成し、サブスクライブすると、最初の GET リクエストによって空のレスポンスが返されます。これは、ポーリング操作がリバランスプロセスを開始してパーティションを割り当てるからです。

3. 手順 2 を繰り返し、Kafka Bridge コンシューマーからメッセージを取得します。

   Kafka Bridge は、レスポンス本文でメッセージの配列 (トピック名、キー、値、パーティション、オフセットの記述) を 200 コードとともに返します。メッセージはデフォルトで最新のオフセットから取得します。

   HTTP/1.1 200 OK
   content-type: application/vnd.kafka.json.v2+json
   #...
   [
     {
       "topic":"bridge-quickstart-topic",
       "key":"my-key",
       "value":"sales-lead-0001",
       "partition":0,
       "offset":0
     },
     {
       "topic":"bridge-quickstart-topic",
       "key":null,
       "value":"sales-lead-0003",
       "partition":0,
       "offset":1
     },
   #...

   注記

   空のレスポンスが返される場合は、トピックおよびパーティションへのメッセージの作成 の説明に従い、コンシューマーに対して追加のレコードを生成し、メッセージの取得を再試行します。

手順

1. quickstart-bridge-topic トピックで、パーティション 0 の特定のオフセットをシークします。

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "offsets": [
           {
               "topic": "bridge-quickstart-topic",
               "partition": 0,
               "offset": 2
           }
       ]
   }'

   リクエストが正常に行われると、Kafka Bridge は 204 コードのみを返します。

2. GET リクエストを records エンドポイントに送信します。

   curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records \
     -H 'accept: application/vnd.kafka.json.v2+json'

   Kafka Bridge は、シークしたオフセットからのメッセージを返します。

3. 同じパーティションの最後のオフセットをシークし、デフォルトのメッセージ取得動作を復元します。今回は、positions/end エンドポイントを使用します。

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions/end \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "partitions": [
           {
               "topic": "bridge-quickstart-topic",
               "partition": 0
           }
       ]
   }'

   リクエストが正常に行われると、Kafka Bridge は別の 204 コードを返します。

手順

1. Kafka Bridge インストールアーカイブで提供される application.properties ファイルを編集します。

   プロパティーファイルを使用して、Kafka および HTTP 関連のプロパティーを指定します。

   1. Kafka コンシューマーおよびプロデューサーに固有のプロパティーなど、標準の Kafka 関連のプロパティーを設定します。

      次の groupId を使用します。

      • kafka.bootstrap.servers: Kafka クラスターへのホスト/ポート接続を定義する。
      • kafka.producer.acks: HTTP クライアントに確認を提供する。

      • kafka.consumer.auto.offset.reset: Kafka のオフセットのリセットを管理する方法を決定する。

        Kafka プロパティーの設定に関する詳細は、Apache Kafka の Web サイトを参照してください。

   2. Kafka クラスターへの HTTP アクセスを有効にするために HTTP 関連のプロパティーを設定します。

      以下に例を示します。

      bridge.id=my-bridge
      http.host=0.0.0.0
      http.port=8080 1
      http.cors.enabled=true 2
      http.cors.allowedOrigins=https://strimzi.io 3
      http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH 4

      1

      8080 番ポートで Kafka Bridge をリッスンするデフォルトの HTTP 設定。

      2

      CORS を有効にするには true に設定します。

      3

      許可される CORS オリジンのコンマ区切りリスト。URL または Java 正規表現を使用できます。

      4

      CORS で許可される HTTP メソッドのコンマ区切りリスト。

2. 設定ファイルを作成します。

手順

1. Kafka Bridge インストールアーカイブで提供される application.properties ファイルを編集します。

   使用するトレースを有効にするには、bridge.tracing プロパティーを使用します。

   OpenTelemetry を有効にする設定例

   bridge.tracing=opentelemetry 1

   1

   OpenTelemetry を有効にするプロパティーは、行の先頭にある # を削除することでコメントが解除されます。

   トレースを有効にすると、Kafka Bridge スクリプトの実行時にトレースを初期化します。

2. 設定ファイルを作成します。

3. トレーシングの環境変数を設定します。

   OpenTelemetry の環境変数

   OTEL_SERVICE_NAME=my-tracing-service 1
   OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 2

   1

   OpenTelemetry トレーサーサービスの名前。

   2

   ポート 4317 でスパンをリッスンする gRPC ベースの OTLP エンドポイント。

4. プロパティーを有効にしてトレース用に Kafka Bridge スクリプトを実行します。

   OpenTelemetry を有効にして Kafka Bridge を実行する

   ./bin/kafka_bridge_run.sh --config-file=<path>/application.properties

   Kafka Bridge の内部コンシューマーとプロデューサーが有効になり、トレースできるようになりました。