第7章 Kafka Bridge

手順

1. examples/kafka-bridge/kafka-bridge.yaml ファイルを編集します。

   apiVersion: kafka.strimzi.io/v1alpha1
   kind: KafkaBridge
   metadata:
     name: quickstart 1
   spec:
     replicas: 1
     bootstrapServers: <cluster-name>-kafka-bootstrap:9092 2
     http:
       port: 8080

   1

   Kafka Bridge のデプロイ時に、デプロイメントの名前およびその他の関連するリソースの名前に -bridge が追加されます。この例では、Kafka Bridge デプロイメントには quickstart-bridge という名前が付けられ、付随する Kafka Bridge サービスには quickstart-bridge-service という名前が付けられます。

   2

   bootstrapServers プロパティーで、Kafka クラスターの名前を <cluster-name> として入力します。

2. Kafka Bridge を OpenShift クラスターにデプロイします。

   oc apply -f examples/kafka-bridge/kafka-bridge.yaml

   quickstart-bridge デプロイメント、サービス、および他の関連リソースが OpenShift クラスターに作成されます。

3. Kafka Bridge が正常にデプロイされたことを確認します。

   oc get deployments

   NAME                             READY   UP-TO-DATE   AVAILABLE   AGE
   quickstart-bridge                  1/1     1            1          34m
   my-cluster-connect                 1/1     1            1          24h
   my-cluster-entity-operator         1/1     1            1          24h
   #...

手順

1. OpenShift クラスターの Pod の名前をリストします。

   oc get pods -o name
   
   pod/kafka-consumer
   # ...
   pod/quickstart-bridge-589d78784d-9jcnr
   pod/strimzi-cluster-operator-76bcf9bc76-8dnfm

2. ポート 8080 で quickstart-bridge Pod に接続します。

   oc port-forward pod/quickstart-bridge-589d78784d-9jcnr 8080:8080 &

   注記

   ローカルマシンのポート 8080 がすでに使用中の場合は、代わりの HTTP ポート (8008 など) を使用します。

手順

1. テキストエディターを使用して、3 つのパーティションがある Kafka トピックの YAML 定義を作成します。

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaTopic
   metadata:
     name: bridge-quickstart-topic
     labels:
       strimzi.io/cluster: <kafka-cluster-name> 1
   spec:
     partitions: 3 2
     replicas: 1
     config:
       retention.ms: 7200000
       segment.bytes: 1073741824

   1

   Kafka Bridge がデプロイされる Kafka クラスターの名前。

   2

   トピックのパーティション数。

2. ファイルを bridge-quickstart-topic.yaml として examples/topic ディレクトリーに保存します。

3. OpenShift クラスターにトピックを作成します。

   oc apply -f examples/topic/bridge-quickstart-topic.yaml

4. 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 トピックのパーティションに送信されます。

5. リクエストが正常に行われると、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-name>-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 コードを返します。