第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