第13章 Kafka Bridge

すべてのリクエストに対して、Content-Type のヘッダーを提出する必要があります。唯一の例外は、POST リクエストボディが空の場合で、Content-Type ヘッダーを追加するとリクエストが失敗します。

コンシューマー操作 (/consumers エンドポイント) とプロデューサー操作 (/topics エンドポイント) では、異なる Content-Type ヘッダーが必要です。

Content-Type: application/vnd.kafka.v2+json

次のセクションで説明されているように、埋め込みデータ形式はコンシューマーごとに設定されます。

POST 要求の本文が空の場合は、Content-Typeを設定しないでください。空の本文を使用して、デフォルト値のコンシューマーを作成できます。

埋め込みデータ形式は、Kafka メッセージが Kafka Bridge によりプロデューサーからコンシューマーに HTTP で送信される際の形式です。サポートされる埋め込みデータ形式には、JSON またはバイナリーの 2 つがサポートされます。

/consumers/groupid エンドポイントを使用してコンシューマーを作成する場合、POST リクエスト本文で JSON またはバイナリーいずれかの埋め込みデータ形式を指定する必要があります。これは、以下の例のようにリクエストボディーの format フィールドで指定します。

{
  "name": "my-consumer",
  "format": "binary", 1
...
}

コンシューマーの埋め込みデータ形式が指定されていない場合は、バイナリー形式が設定されます。

コンシューマーの作成時に指定する埋め込みデータ形式は、コンシューマーが消費する Kafka メッセージのデータ形式と一致する必要があります。

バイナリー埋め込みデータ形式を指定する場合は、以降のプロデューサー要求で、要求本文にバイナリーデータが Base64 でエンコードされた文字列として含まれる必要があります。たとえば、POST リクエストを /topics/topicname エンドポイントに対して作成してメッセージを送信する場合、value は Base64 でエンコードする必要があります。

{
  "records": [
    {
      "key": "my-key",
      "value": "ZWR3YXJkdGhldGhyZWVsZWdnZWRjYXQ="
    },
  ]
}

プロデューサー要求には、埋め込みデータ形式に対応する Content-Type ヘッダーも含まれる必要があります (例: Content-Type: application/vnd.kafka.binary.v2+json)。

/topics エンドポイントを使用してメッセージを送信する場合は、records パラメーターの要求本文にメッセージペイロードを入力します。

records パラメーターには、以下のオプションフィールドを含めることができます。

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

コンシューマーを作成したら、以降のすべての GET 要求には Accept ヘッダーが以下のような形式で含まれる必要があります。

Accept: application/vnd.kafka.embedded-data-format.v2+json

embedded-data-format は json または binary です。

たとえば、サブスクライブされたコンシューマーのレコードを JSON 埋め込みデータ形式で取得する場合、この Accept ヘッダーが含まれるようにします。

Accept: application/vnd.kafka.json.v2+json