1. Home
  2. Platform
  3. Confluent REST API

Confluent REST Proxy API リファレンス

コンテンツタイプ

REST Proxy では、リクエストと応答の両方にコンテンツタイプを使用し、以下のデータプロパティを示します。

  • シリアル化のフォーマット: json

  • API バージョン（例: v2 または v3）

  • 組み込みフォーマット: json、binary、avro、protobuf、jsonschema

    重要

    REST Proxy v2 以降では、jsonschema および protobuf の組み込み型がサポートされています。

REST Proxy は、Avro®、 JSON スキーマ、および Protobuf シリアル化フォーマットをサポートしています。REST Proxy API のバージョンは v2 と v3 です。

組み込みフォーマットは、生成または消費するデータのフォーマットです。これらのフォーマットは、シリアル化フォーマットでリクエストまたは応答に組み込まれます。たとえば、json でシリアル化されたリクエストに binary データを指定できます。この場合、データは base64 エンコード文字列として指定します。

  • v2 の場合、コンテンツタイプは application/vnd.kafka.binary.v2+json になります。
  • v3 の場合、コンテンツタイプは application/json になります。

データが JSON の場合、json を組み込みフォーマットとして使用し、それを直接組み込むことができます。

  • v2 の場合、コンテンツタイプは application/vnd.kafka.json.v2+json になります。
  • v3 の場合、コンテンツタイプは application/json になります。

avro、protobuf、および jsonschema 組み込み型を使用すると、JSON フォーマットのデータをスキーマ（またはスキーマ ID）とともにリクエストに直接組み込むことができます。これらの型は Schema Registry を使用し、データおよびペイロードに加えてスキーマの ID がシリアル化されます。

  • Avro コンテンツタイプは application/vnd.kafka.avro.v2+json です。
  • Protobuf コンテンツタイプは application/vnd.kafka.protobuf.v2+json です。
  • JSON スキーマのコンテンツタイプは application/vnd.kafka.jsonschema.v2+json です。

コンテンツタイプのフォーマットは以下のとおりです。:

application/vnd.kafka[.embedded_format].[api_version]+[serialization_format]

詳細については、「Schema Registry API リファレンス」を参照してください。

メッセージが組み込まれていない場合、組み込みフォーマットを省略できます（つまり、メタデータのリクエストには application/vnd.kafka.v2+json を使用できます）。v2 の場合に推奨されるコンテンツタイプは application/vnd.kafka.[embedded_format].v2+json です。ただし、コンテンツタイプをより抽象的にして指定することもできます。たとえば、使用すべき API バージョン要件を指定せずに application/vnd.kafka+json （最新の安定版が使用されます）としたり、application/json や application/octet-stream のようにしたりすることができます。後者の 2 つは、あくまで互換性と使用の容易性を意図してサポートされています。いずれの場合も、組み込みフォーマットを省略すると、binary と見なされます。これらの抽象的な値の使用は許可されていますが、将来のバージョンとの互換性を確保するためには、推奨されるコンテンツタイプをリクエストに指定するとともに、応答のコンテンツタイプをチェックする "必要があります"。

実際のリクエストでは、フォーマットとバージョンの情報を HTTP の Accept ヘッダーで、できるだけ具体的に指定する 必要があります。

v2 の場合は、フォーマットとバージョンを次のように指定できます。

Accept: application/vnd.kafka.v2+json

v3 の場合は、バージョンを指定しないでください。最新のバージョン（v3）が使用されます。

Accept: application/json

サーバーでは、コンテンツネゴシエーションもサポートされるため、重み付けされた複数の優先度指定を含めることができます。:

Accept: application/vnd.kafka.v2+json; q=0.9, application/json; q=0.5

この方法は、新しいバージョンの API が推奨されるものの、そのバージョンの API が利用できるかどうかが定かでない場合などに活用できます。

参考

「REST API の使用例（curl）」では、curl を使用してコマンドラインから API をテストする方法について説明しています。

エラー

すべての API エンドポイントでは、エラーを表す HTTP ステータス（400 番台または 500 番台のステータス）がリクエストから返された場合、標準のエラーメッセージフォーマットが使用されます。たとえば、リクエストエンティティの必須フィールドに漏れがあると、次のような応答が生成されることがあります。

HTTP/1.1 422 Unprocessable Entity
Content-Type: application/vnd.kafka.v3+json

{
    "error_code": 422,
    "message": "records may not be empty"
}

ステータスコードをチェックすることをお勧めしますが、DELETE API 以外であれば、API 呼び出しのレスポンスを解析し、error_code フィールドの存在をチェックすることによってエラーを検出してもかまいません。

一部のエラーコードは API 全体で頻繁に使用されるため、これらを処理する汎用コードが必要になる可能性が高くなりますが、その他のほとんどのエラーコードはリクエストごとに処理する必要があります。

ANY /
ステータスコード:
  • 401 Unauthorized --
    • エラーコード 40101 -- Kafka 認証エラー。
  • 403 Forbidden --
    • エラーコード 40301 -- Kafka 認可エラー。
  • 404 Not Found --
    • エラーコード 40401 -- トピックが見つかりません。
    • エラーコード 40402 -- パーティションが見つかりません。
  • 422 Unprocessable Entity -- リクエストのペイロードが正しくフォーマットされていないか、セマンティクスエラーが含まれています
  • 500 Internal Server Error --
    • Error code 50001 -- Zookeeper error.
    • Error code 50002 -- Kafka error.
    • Error code 50003 -- Retriable Kafka error. Although the operation failed, it's possible that retrying the request will be successful.
    • エラーコード 50101 -- 指定されたブローカーについてのみ SSL エンドポイントが見つかりましたが、呼び出された API については SSL はまだサポートされていません。

REST Proxy API v2

ちなみに

API エンドポイントをコマンドラインからテストする方法については、「REST API の使用例（curl）」を参照してください。

トピック

トピックリソースは、Kafka クラスターのトピックとその現在のステートに関する情報を提供します。また、特定のトピックに対して POST リクエストを行うことでメッセージを生成することもできます。

GET /topics

Kafka のトピックのリストを取得します。

レスポンスの JSON オブジェクト:
 
  • topics (array) -- トピック名のリスト

リクエストの例 :

GET /topics HTTP/1.1
Host: kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json

応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

["topic1", "topic2"]
GET /topics/(string: topic_name)

特定のトピックに関するメタデータを取得します。

パラメーター:
  • topic_name (string) -- メタデータを取得するトピックの名前
レスポンスの JSON オブジェクト:
 
  • name (string) -- トピックの名前
  • configs (map) -- トピックごとの構成のオーバーライド
  • partitions (array) -- このトピックのパーティションのリスト
  • partitions[i].partition (int) -- このパーティションの ID
  • partitions[i].leader (int) -- このパーティションのリーダーのブローカー ID
  • partitions[i].replicas (array) -- リーダーを含むこのパーティションのレプリカのリスト
  • partitions[i].replicas[j].broker (array) -- レプリカのブローカー ID
  • partitions[i].replicas[j].leader (boolean) -- このレプリカがパーティションのリーダーである場合は true
  • partitions[i].replicas[j].in_sync (boolean) -- このレプリカが現在リーダーと同期している場合は true
ステータスコード:
  • 404 Not Found --
    • エラーコード 40401 -- トピックが見つかりません

リクエストの例 :

GET /topics/test HTTP/1.1
Accept: application/vnd.kafka.v2+json

応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "name": "test",
  "configs": {
     "cleanup.policy": "compact"
  },
  "partitions": [
    {
      "partition": 1,
      "leader": 1,
      "replicas": [
        {
          "broker": 1,
          "leader": true,
          "in_sync": true,
        },
        {
          "broker": 2,
          "leader": false,
          "in_sync": true,
        }
      ]
    },
    {
      "partition": 2,
      "leader": 2,
      "replicas": [
        {
          "broker": 1,
          "leader": false,
          "in_sync": true,
        },
        {
          "broker": 2,
          "leader": true,
          "in_sync": true,
        }
      ]
    }
  ]
}
POST /topics/(string: topic_name)

任意でメッセージのキーまたはパーティションを指定して、トピックにメッセージを生成します。パーティションが指定されていない場合は、キーのハッシュに基づいてパーティションが選択されます。キーが指定されていない場合、メッセージごとにラウンドロビン方式でパーティションが選択されます。

avro、protobuf、および jsonschema 組み込みフォーマットの場合は、スキーマに関する情報を指定し、Schema Registry にアクセスする URL（ schema.registry.url ）を使用して REST Proxy を構成する必要があります。スキーマは、文字列としてエンコードされた完全なスキーマとして指定できます。または、最初のリクエストの後に、最初の応答で返されたスキーマ ID として指定できます。

パラメーター:
  • topic_name (string) -- メッセージを生成する先のトピックの名前
リクエストの JSON オブジェクト:
 
  • key_schema (string) -- 文字列としてエンコードされた完全なスキーマ（たとえば、Avro データ用にシリアル化された JSON）
  • key_schema_id (int) -- 同じスキーマを使用する前のリクエストが返した ID。この ID は、レジストリ内のスキーマの ID に対応します。
  • value_schema (string) -- 文字列としてエンコードされた完全なスキーマ（たとえば、Avro データ用にシリアル化された JSON）
  • value_schema_id (int) -- 同じスキーマを使用する前のリクエストが返した ID。この ID は、レジストリ内のスキーマの ID に対応します。
リクエスト（オブジェクトの JSON 配列）:
 
  • records -- トピックに対して生成するレコードのリスト
  • records[i].key (object) -- 組み込みフォーマットに従ってフォーマットされたメッセージキー。キーを省略する場合は null（オプション）
  • records[i].value (object) -- 組み込みフォーマットに従ってフォーマットされたメッセージ値
  • records[i].partition (int) -- メッセージを保存するパーティション（オプション）
レスポンスの JSON オブジェクト:
 
  • key_schema_id (int) -- キーの生成に使用されるスキーマの ID。キーが使用されなかった場合は null
  • value_schema_id (int) -- 値を生成するために使用されるスキーマの ID。
応答（オブジェクトの JSON 配列）:
 
  • offsets (object) -- メッセージのパブリッシュ先のパーティションとオフセットのリスト
  • offsets[i].partition (int) -- メッセージのパブリッシュ先のパーティション。メッセージのパブリッシュが失敗した場合は null
  • offsets[i].offset (long) -- メッセージのオフセット。メッセージのパブリッシュが失敗した場合は null
  • offsets[i].error_code (long) -- この操作が失敗した理由を分類するエラーコード。成功した場合は null。* 1 - 再試行不可な Kafka 例外 * 2 - 再試行可能な Kafka 例外。再試行すると、メッセージが正常に送信される場合があります
  • offsets[i].error (string) -- 操作が失敗した理由を示すエラーメッセージ。成功した場合は null
ステータスコード:
  • 404 Not Found --
    • エラーコード 40401 -- トピックが見つかりません
  • 422 Unprocessable Entity --
    • エラーコード 42201 -- リクエストにキーが含まれており、スキーマを必要とするフォーマットを使用していますが、key_schema または key_schema_id フィールドが含まれていません
    • エラーコード 42202 -- リクエストに値が含まれており、スキーマを必要とするフォーマットを使用していますが、value_schema または value_schema_id フィールドが含まれていません
    • エラーコード 42205 -- リクエストに無効なスキーマが含まれています。
  • 408 Request Timeout --
    • エラーコード 40801 -- スキーマの登録または検索に失敗しました。

バイナリリクエストの例:

POST /topics/test HTTP/1.1
Host: kafkaproxy.example.com
Content-Type: application/vnd.kafka.binary.v2+json
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

{
  "records": [
    {
      "key": "a2V5",
      "value": "Y29uZmx1ZW50"
    },
    {
      "value": "a2Fma2E=",
      "partition": 1
    },
    {
      "value": "bG9ncw=="
    }
  ]
}

バイナリ応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "key_schema_id": null,
  "value_schema_id": null,
  "offsets": [
    {
      "partition": 2,
      "offset": 100
    },
    {
      "partition": 1,
      "offset": 101
    },
    {
      "partition": 2,
      "offset": 102
    }
  ]
}

Avro リクエストの例 :

POST /topics/test HTTP/1.1
Host: kafkaproxy.example.com
Content-Type: application/vnd.kafka.avro.v2+json
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

{
  "value_schema": "{\"name\":\"int\",\"type\": \"int\"}",
  "records": [
    {
      "value": 12
    },
    {
      "value": 24,
      "partition": 1
    }
  ]
}

Avro 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "key_schema_id": null,
  "value_schema_id": 32,
  "offsets": [
    {
      "partition": 2,
      "offset": 103
    },
    {
      "partition": 1,
      "offset": 104
    }
  ]
}

JSON リクエストの例 :

POST /topics/test HTTP/1.1
Host: kafkaproxy.example.com
Content-Type: application/vnd.kafka.json.v2+json
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

{
  "records": [
    {
      "key": "somekey",
      "value": {"foo": "bar"}
    },
    {
      "value": [ "foo", "bar" ],
      "partition": 1
    },
    {
      "value": 53.5
    }
  ]
}

JSON 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "key_schema_id": null,
  "value_schema_id": null,
  "offsets": [
    {
      "partition": 2,
      "offset": 100
    },
    {
      "partition": 1,
      "offset": 101
    },
    {
      "partition": 2,
      "offset": 102
    }
  ]
}

パーティション

パーティションリソースは、現在のリーダーや各パーティションのレプリカなど、パーティションごとのメタデータを提供します。また、GET リクエストと POST リクエストを使用して、1 つのパーティションに対して、メッセージを消費したり生成したりすることもできます。

GET /topics/(string: topic_name)/partitions

トピックのパーティションのリストを取得します。

パラメーター:
  • topic_name (string) -- トピックの名前
応答（オブジェクトの JSON 配列）:
 
  • partition (int) -- パーティションの ID
  • leader (int) -- このパーティションのリーダーのブローカー ID
  • replicas (array) -- このパーティションのレプリカとして機能するブローカーのリスト
  • replicas[i].broker (int) -- レプリカのブローカー ID
  • replicas[i].leader (boolean) -- このブローカーがパーティションのリーダーである場合は true
  • replicas[i].in_sync (boolean) -- このレプリカがリーダーと同期している場合は true
ステータスコード:
  • 404 Not Found --
    • エラーコード 40401 -- トピックが見つかりません

リクエストの例 :

GET /topics/test/partitions HTTP/1.1
Host: kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

[
  {
    "partition": 1,
    "leader": 1,
    "replicas": [
      {
        "broker": 1,
        "leader": true,
        "in_sync": true,
      },
      {
        "broker": 2,
        "leader": false,
        "in_sync": true,
      },
      {
        "broker": 3,
        "leader": false,
        "in_sync": false,
      }
    ]
  },
  {
    "partition": 2,
    "leader": 2,
    "replicas": [
      {
        "broker": 1,
        "leader": false,
        "in_sync": true,
      },
      {
        "broker": 2,
        "leader": true,
        "in_sync": true,
      },
      {
        "broker": 3,
        "leader": false,
        "in_sync": false,
      }
    ]
  }
]
GET /topics/(string: topic_name)/partitions/(int: partition_id)

トピック内の 1 つのパーティションに関するメタデータを取得します。

パラメーター:
  • topic_name (string) -- トピックの名前
  • partition_id (int) -- 検査するパーティションの ID
レスポンスの JSON オブジェクト:
 
  • partition (int) -- パーティションの ID
  • leader (int) -- このパーティションのリーダーのブローカー ID
  • replicas (array) -- このパーティションのレプリカとして機能するブローカーのリスト
  • replicas[i].broker (int) -- レプリカのブローカー ID
  • replicas[i].leader (boolean) -- このブローカーがパーティションのリーダーである場合は true
  • replicas[i].in_sync (boolean) -- このレプリカがリーダーと同期している場合は true
ステータスコード:
  • 404 Not Found --
    • エラーコード 40401 -- トピックが見つかりません
    • エラーコード 40402 -- パーティションが見つかりません

リクエストの例 :

GET /topics/test/partitions/1 HTTP/1.1
Host: kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "partition": 1,
  "leader": 1,
  "replicas": [
    {
      "broker": 1,
      "leader": true,
      "in_sync": true,
    },
    {
      "broker": 2,
      "leader": false,
      "in_sync": true,
    },
    {
      "broker": 3,
      "leader": false,
      "in_sync": false,
    }
  ]
}
GET /topics/(string: topic_name)/partitions/(int: partition_id)/offsets

このトピックのパーティションのオフセットのサマリーを取得します。

パラメーター:
  • topic_name (string) -- トピックの名前
  • partition_id (int) -- 検査するパーティションの ID
レスポンスの JSON オブジェクト:
 
  • beginning_offset (int) -- このパーティションの最初のオフセット
  • end_offset (int) -- このパーティションの最後のオフセット
ステータスコード:
  • 404 Not Found --
    • エラーコード 40401 -- トピックが見つかりません
    • エラーコード 40402 -- パーティションが見つかりません

リクエストの例 :

GET /topics/test/partitions/1/offsets HTTP/1.1
Host: kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "beginning_offset": 10,
  "end_offset": 50,
}
POST /topics/(string: topic_name)/partitions/(int: partition_id)

トピックの 1 つのパーティションにメッセージを生成します。Avro、JSON スキーマ、および Protobuf 組み込みフォーマットの場合、スキーマに関する情報を指定する必要があります。このスキーマは、文字列としてエンコードされた完全なスキーマとして指定できます。または、最初のリクエストの後に、最初の応答で返されたスキーマ ID として指定できます。

パラメーター:
  • topic_name (string) -- メッセージを生成する先のトピック
  • partition_id (int) -- メッセージを生成する先のパーティション
リクエストの JSON オブジェクト:
 
  • key_schema (string) -- 文字列としてエンコードされた完全なスキーマ（たとえば、Avro データ用にシリアル化された JSON）
  • key_schema_id (int) -- 同じスキーマを使用する前のリクエストが返した ID。この ID は、レジストリ内のスキーマの ID に対応します。
  • value_schema (string) -- 文字列としてエンコードされた完全なスキーマ（たとえば、Avro データ用にシリアル化された JSON）
  • value_schema_id (int) -- 同じスキーマを使用する前のリクエストが返した ID。この ID は、レジストリ内のスキーマの ID に対応します。
  • records -- パーティションに対して生成するレコードのリスト。
リクエスト（オブジェクトの JSON 配列）:
 
  • records[i].key (object) -- 組み込みフォーマットに従ってフォーマットされたメッセージキー。キーを省略する場合は null（オプション）
  • records[i].value (object) -- 組み込みフォーマットに従ってフォーマットされたメッセージ値
レスポンスの JSON オブジェクト:
 
  • key_schema_id (int) -- キーの生成に使用されるスキーマの ID。キーが使用されなかった場合は null
  • value_schema_id (int) -- 値を生成するために使用されるスキーマの ID。
応答（オブジェクトの JSON 配列）:
 
  • offsets (object) -- メッセージのパブリッシュ先のパーティションとオフセットのリスト
  • offsets[i].partition (int) -- メッセージのパブリッシュ先のパーティション。これは partition_id パラメーターと同じで、トピックに対する生成からの応答との一貫性を維持するためだけに提供されます。
  • offsets[i].offset (long) -- メッセージのオフセット
  • offsets[i].error_code (long) -- この操作が失敗した理由を分類するエラーコード。成功した場合は null。* 1 - 再試行不可な Kafka 例外 * 2 - 再試行可能な Kafka 例外。再試行すると、メッセージが正常に送信される場合があります
  • offsets[i].error (string) -- 操作が失敗した理由を示すエラーメッセージ。成功した場合は null
ステータスコード:
  • 404 Not Found --
    • エラーコード 40401 -- トピックが見つかりません
    • エラーコード 40402 -- パーティションが見つかりません
  • 422 Unprocessable Entity --
    • エラーコード 42201 -- リクエストにキーが含まれており、スキーマを必要とするフォーマットを使用していますが、key_schema または key_schema_id フィールドが含まれていません
    • エラーコード 42202 -- リクエストに値が含まれており、スキーマを必要とするフォーマットを使用していますが、value_schema または value_schema_id フィールドが含まれていません
    • エラーコード 42205 -- リクエストに無効なスキーマが含まれています。

バイナリリクエストの例:

POST /topics/test/partitions/1 HTTP/1.1
Host: kafkaproxy.example.com
Content-Type: application/vnd.kafka.binary.v2+json
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

{
  "records": [
    {
      "key": "a2V5",
      "value": "Y29uZmx1ZW50"
    },
    {
      "value": "a2Fma2E="
    }
  ]
}

バイナリ応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "key_schema_id": null,
  "value_schema_id": null,
  "offsets": [
    {
      "partition": 1,
      "offset": 100,
    },
    {
      "partition": 1,
      "offset": 101,
    }
  ]
}

Avro リクエストの例 :

POST /topics/test/partitions/1 HTTP/1.1
Host: kafkaproxy.example.com
Content-Type: application/vnd.kafka.avro.v2+json
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

{
  "value_schema": "{\"name\":\"int\",\"type\": \"int\"}"
  "records": [
    {
      "value": 25
    },
    {
      "value": 26
    }
  ]
}

Avro 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "key_schema_id": null,
  "value_schema_id": 32,
  "offsets": [
    {
      "partition": 1,
      "offset": 100,
    },
    {
      "partition": 1,
      "offset": 101,
    }
  ]
}

JSON リクエストの例 :

POST /topics/test/partitions/1 HTTP/1.1
Host: kafkaproxy.example.com
Content-Type: application/vnd.kafka.json.v2+json
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

{
  "records": [
    {
      "key": "somekey",
      "value": {"foo": "bar"}
    },
    {
      "value": 53.5
    }
  ]
}

JSON 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "key_schema_id": null,
  "value_schema_id": null,
  "offsets": [
    {
      "partition": 1,
      "offset": 100,
    },
    {
      "partition": 1,
      "offset": 101,
    }
  ]
}

PROTOBUF リクエストの例 :

POST /topics/test/partitions/1 HTTP/1.1
Content-Type: application/vnd.kafka.protobuf.v2+json
Accept: application/vnd.kafka.v2+json, application/json

{
  "value_schema": "syntax=\"proto3\"; message Foo { string f1 = 1; }"
  "records": [{"value": {"f1": "foo"}}]
}

PROTOBUF 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "key_schema_id": null,
  "value_schema_id": 32,
  "offsets": [
    {
      "partition": 1,
      "offset": 100,
    },
    {
      "partition": 1,
      "offset": 101,
    }
  ]
}

JSONSCHEMA リクエストの例 :

POST /topics/test/partitions/1 HTTP/1.1
Content-Type: application/vnd.kafka.jsonschema.v2+json
Accept: application/vnd.kafka.v2+json, application/json

{
  "value_schema": "{\"type\":\"object\",\"properties\":{\"f1\":{\"type\":\"string\"}}}",
  "records": [{"value": {"f1": "bar"}}]
}

JSONSCHEMA 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "key_schema_id": null,
  "value_schema_id": 32,
  "offsets": [
    {
      "partition": 1,
      "offset": 100,
    },
    {
      "partition": 1,
      "offset": 101,
    }
  ]
}

コンシューマー

コンシューマーリソースによって、コンシューマーグループの現在のステートを確認できます。その結果、コンシューマグループにコンシューマーを作成したり、トピックおよびパーティションからメッセージを消費したりできます。REST Proxy は、Kafka にシリアル化されたフォーマットで保存されたデータを JSON 互換性のある組み込みフォーマットに変換することができます。以下のフォーマットがサポートされています。

  • 未加工のバイナリデータは base64 文字列としてエンコードされる
  • Avro データは組み込みデータに変換される
  • JSON オブジェクト（JSON は直接組み込まれる）
  • Protobuf
  • JSON スキーマ

コンシューマーはステートフルであるため、REST API で作成されたコンシューマーインスタンスはすべて特定の REST Proxy インスタンスに関連付けられます。インスタンスが作成されるときに完全な URL が提供され、以降のリクエストの作成に使用されます。返された URL を以降のコンシューマーリクエストに使用していない場合、コンシューマーインスタンスが見つからないため、"404" エラーが発生します。REST Proxy インスタンスがシャットダウンする場合、インスタンスが終了する前にすべてのコンシューマーをクリーンに破棄しようとします。

POST /consumers/(string: group_name)

コンシューマーグループに新規コンシューマーインスタンスを作成します。format パラメーターは、Kafka からのデータの逆シリアル化と、このコンシューマーに対して実行される後続の読み取り API リクエストの Accept ヘッダーで使用する "必須の" コンテンツタイプを制御します。たとえば、作成リクエストでフォーマットに avro が指定されている場合、後続の読み取りリクエストでは Accept: application/vnd.kafka.avro.v2+json を使用する必要があります。

コンシューマーはステートフルで、特定の REST Proxy インスタンスに関連付けられているため、応答にはホストを含む URL が含まれることに注意してください。このセクションの以降の例では、この特定の REST Proxy インスタンスに Host ヘッダーを使用します。

パラメーター:
  • group_name (string) -- 結合するコンシューマーグループの名前
リクエストの JSON オブジェクト:
 
  • name (string) -- コンシューマーインスタンスの名前。コンシューマーの URL で使用されます。これは、少なくともリクエストを処理する REST Proxy プロセス内で一意である必要があります。省略すると、自動的に生成された ID が使用されます。ほとんどのユースケースでは、自動的に生成された名前を使用することが推奨されます。
  • format (string) -- メッセージを JSON 互換性のある形式に変換するために使用される、消費されたメッセージのフォーマット。指定可能な値は、"binary"、"avro"、"json"、"jsonschema"、および protobuf です。指定しないと、デフォルトで "binary" になります。
  • auto.offset.reset (string) -- コンシューマーの auto.offset.reset を設定します。
  • auto.commit.enable (string) -- コンシューマーの auto.commit.enable を設定します。
  • fetch.min.bytes (string) -- このコンシューマーに固有の fetch.min.bytes を設定します。
  • consumer.request.timeout.ms (string) -- このコンシューマーに固有の consumer.request.timeout.ms を設定します。この設定は、最大リクエストサイズにまだ達していない場合に、リクエストのメッセージを待機する最大合計時間を制御します。コンシューマーからブローカーへの基礎的な接続には影響しません。デフォルト値は REST Proxy 構成ファイルから取得されます。
レスポンスの JSON オブジェクト:
 
  • instance_id (string) -- このグループ内のコンシューマーインスタンスの一意の ID。
  • base_uri (string) -- このコンシューマーインスタンスに対する後続のリクエストの URI を構築するために使用されるベース URI。http://hostname:port/consumers/consumer_group/instances/instance_id という形式になります。
ステータスコード:
  • 409 Conflict --
    • エラーコード 40902 -- 指定された名前のコンシューマーインスタンスが既に存在します。
  • 422 Unprocessable Entity --
    • エラーコード 42204 -- コンシューマー構成が無効です。リクエストに指定された設定の 1 つに無効な値が含まれています。

リクエストの例 :

POST /consumers/testgroup/ HTTP/1.1
Host: kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json


{
  "name": "my_consumer",
  "format": "binary",
  "auto.offset.reset": "earliest",
  "auto.commit.enable": "false"
}

応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "instance_id": "my_consumer",
  "base_uri": "http://proxy-instance.kafkaproxy.example.com/consumers/testgroup/instances/my_consumer"
}

PROTOBUF リクエストの例 :

POST /consumers/testgroup/ HTTP/1.1
Host: kafkaproxy.example.com
Content-Type: application/vnd.kafka.protobuf.v2+json


{
  "name": "my_consumer",
  "format": "protobuf",
  "auto.offset.reset": "earliest",
  "auto.commit.enable": "false"
}

PROTOBUF 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.protobuf.v2+json

{
  "instance_id": "my_consumer",
  "base_uri": "http://proxy-instance.kafkaproxy.example.com/consumers/my_protobuf_consumer"
}

JSONSCHEMA リクエストの例 :

POST /consumers/testgroup/ HTTP/1.1
Host: kafkaproxy.example.com
Content-Type: application/vnd.kafka.jsonschema.v2+json


{
  "name": "my_consumer",
  "format": "jsonschema",
  "auto.offset.reset": "earliest",
  "auto.commit.enable": "false"
}

JSONSCHEMA 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.jsonschema.v2+json

{
  "instance_id": "my_consumer",
  "base_uri": "http://proxy-instance.kafkaproxy.example.com/consumers/my_jsonschema_consumer"
}
DELETE /consumers/(string: group_name)/instances/(string: instance)

コンシューマーインスタンスを破棄します。

コンシューマーインスタンスを保持する特定の REST Proxy インスタンスに対して "必ず" 行うリクエストです。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません

リクエストの例 :

DELETE /consumers/testgroup/instances/my_consumer HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json

応答の例 :

HTTP/1.1 204 No Content
POST /consumers/(string: group_name)/instances/(string: instance)/offsets

コンシューマーのオフセットのリストをコミットします。post の本文が空の場合、コンシューマーインスタンスによってフェッチされたすべてのレコードがコミットされます。

コンシューマーインスタンスを保持する特定の REST Proxy インスタンスに対して "必ず" 行うリクエストです。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
リクエスト（オブジェクトの JSON 配列）:
 
  • offsets -- パーティション用にコミットするオフセットのリスト
  • offsets[i].topic (string) -- トピックの名前
  • offsets[i].partition (int) -- パーティション ID
  • offset -- コミットするオフセット
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません

リクエストの例 :

POST /consumers/testgroup/instances/my_consumer/offsets HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json

{
  "offsets": [
    {
      "topic": "test",
      "partition": 0,
      "offset": 20
    },
    {
      "topic": "test",
      "partition": 1,
      "offset": 30
    }
  ]
}
GET /consumers/(string: group_name)/instances/(string: instance)/offsets

指定したパーティションについて、最後にコミットされたオフセット（コミットがこの処理で発生したか別のプロセスで発生したか）を取得します。

コンシューマーインスタンスを保持する特定の REST Proxy インスタンスに対して "必ず" 行うリクエストです。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
リクエスト（オブジェクトの JSON 配列）:
 
  • partitions -- 最後にコミットされたオフセットを検索するパーティションのリスト
  • partitions[i].topic (string) -- トピックの名前
  • partitions[i].partition (int) -- パーティション ID
応答（オブジェクトの JSON 配列）:
 
  • offsets -- コミットされたオフセットのリスト
  • offsets[i].topic (string) -- オフセットがコミットされたトピックの名前
  • offsets[i].partition (int) -- オフセットがコミットされたパーティション ID
  • offsets[i].offset (int) -- コミットされたオフセット
  • offsets[i].metadata (string) -- コミットされたオフセットのメタデータ
ステータスコード:
  • 404 Not Found --
    • エラーコード 40402 -- パーティションが見つかりません
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません

リクエストの例 :

GET /consumers/testgroup/instances/my_consumer/offsets HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    }

  ]
}

応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{"offsets":
 [
  {
    "topic": "test",
    "partition": 0,
    "offset": 21,
    "metadata":""
  },
  {
    "topic": "test",
    "partition": 1,
    "offset": 31,
    "metadata":""
  }
 ]
}
POST /consumers/(string: group_name)/instances/(string: instance)/subscription

指定したトピックのリストまたはトピックパターンをサブスクライブし、動的に割り当てられたパーティションを取得します。以前のサブスクリプションが存在する場合は、最新のサブスクリプションに置き換えられます。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
リクエスト（オブジェクトの JSON 配列）:
 
  • topics -- サブスクライブするトピックのリスト
  • topics[i].topic (string) -- トピックの名前
リクエストの JSON オブジェクト:
 
  • topic_pattern (string) -- REGEX パターン。topics_pattern フィールドと topics フィールドは同時に使用できません。
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません
  • 409 Conflict --
    • エラーコード 40903 -- トピック、パーティション、およびパターンへのサブスクリプションは同時にできません。

リクエストの例 :

POST /consumers/testgroup/instances/my_consumer/subscription HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json

{
  "topics": [
    "test1",
    "test2"
  ]
}

応答の例 :

HTTP/1.1 204 No Content

リクエストの例 :

POST /consumers/testgroup/instances/my_consumer/subscription HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json

{
  "topic_pattern": "test.*"
}

応答の例 :

HTTP/1.1 204 No Content
GET /consumers/(string: group_name)/instances/(string: instance)/subscription

現在サブスクライブしているトピックのリストを取得します。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
応答（オブジェクトの JSON 配列）:
 
  • topics -- サブスクライブしているトピックのリスト
  • topics[i] (string) -- トピックの名前
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません

リクエストの例 :

GET /consumers/testgroup/instances/my_consumer/subscription HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "topics": [
    "test1",
    "test2"
  ]
}
DELETE /consumers/(string: group_name)/instances/(string: instance)/subscription

現在サブスクライブしているトピックのサブスクライブを解除します。

コンシューマーインスタンスを保持する特定の REST Proxy インスタンスに対して "必ず" 行うリクエストです。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません

リクエストの例 :

DELETE /consumers/testgroup/instances/my_consumer/subscription HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

応答の例 :

HTTP/1.1 204 No Content
POST /consumers/(string: group_name)/instances/(string: instance)/assignments

このコンシューマーにパーティションのリストを手動で割り当てます。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
リクエスト（オブジェクトの JSON 配列）:
 
  • partitions -- このコンシューマーに割り当てるパーティションのリスト
  • partitions[i].topic (string) -- トピックの名前
  • partitions[i].partition (int) -- パーティション ID
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません
  • 409 Conflict --
    • エラーコード 40903 -- トピック、パーティション、およびパターンへのサブスクリプションは同時にできません。

リクエストの例 :

POST /consumers/testgroup/instances/my_consumer/assignments HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json

{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    }

  ]
}

応答の例 :

HTTP/1.1 204 No Content
GET /consumers/(string: group_name)/instances/(string: instance)/assignments

このコンシューマーに現在手動で割り当てられているパーティションのリストを取得します。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
応答（オブジェクトの JSON 配列）:
 
  • partitions -- このコンシューマーに手動で割り当てられているパーティションのリスト
  • partitions[i].topic (string) -- トピックの名前
  • partitions[i].partition (int) -- パーティション ID
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません

リクエストの例 :

GET /consumers/testgroup/instances/my_consumer/assignments HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    }

  ]
}
POST /consumers/(string: group_name)/instances/(string: instance)/positions

コンシューマーが次にフェッチするレコードセットに使用するフェッチオフセットをオーバーライドします。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
リクエスト（オブジェクトの JSON 配列）:
 
  • offsets -- オフセットのリスト
  • offsets[i].topic (string) -- オフセットがコミットされたトピックの名前
  • offsets[i].partition (int) -- パーティション ID
  • offsets[i].offset (int) -- 次にフェッチするレコードセットのオフセットをシークします。
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません

リクエストの例 :

POST /consumers/testgroup/instances/my_consumer/positions HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json


{
  "offsets": [
    {
      "topic": "test",
      "partition": 0,
      "offset": 20
    },
    {
      "topic": "test",
      "partition": 1,
      "offset": 30
    }
  ]
}

応答の例 :

HTTP/1.1 204 No Content
POST /consumers/(string: group_name)/instances/(string: instance)/positions/beginning

指定した各パーティションの最初のオフセットをシークします。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
リクエスト（オブジェクトの JSON 配列）:
 
  • partitions -- パーティションのリスト
  • partitions[i].topic (string) -- トピックの名前
  • partitions[i].partition (int) -- パーティション ID
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません

リクエストの例 :

POST /consumers/testgroup/instances/my_consumer/positions/beginning HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json

{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    }

  ]
}

応答の例 :

HTTP/1.1 204 No Content
POST /consumers/(string: group_name)/instances/(string: instance)/positions/end

指定した各パーティションの最後のオフセットをシークします。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
リクエスト（オブジェクトの JSON 配列）:
 
  • partitions -- パーティションのリスト
  • partitions[i].topic (string) -- トピックの名前
  • partitions[i].partition (int) -- パーティション ID
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません

リクエストの例 :

POST /consumers/testgroup/instances/my_consumer/positions/end HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Content-Type: application/vnd.kafka.v2+json

{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    }

  ]
}

応答の例 :

HTTP/1.1 204 No Content
GET /consumers/(string: group_name)/instances/(string: instance)/records

サブスクライブ API または割り当て API のいずれかを使用して指定されたトピックまたはパーティションのデータをフェッチします。

このリクエストによって返される組み込みデータのフォーマットは、最初のコンシューマーインスタンス作成リクエストで指定されたフォーマットによって決まり、Accept ヘッダーのフォーマットと一致する必要があります。一致しない場合、エラーコード 40601 が発生します。

コンシューマーインスタンスを保持する特定の REST Proxy インスタンスに対して "必ず" 行うリクエストです。

パラメーター:
  • group_name (string) -- コンシューマーグループの名前
  • instance (string) -- コンシューマーインスタンスの ID
クエリのパラメーター:
  • timeout -- REST Proxy がレコードのフェッチに費やす最大時間（ミリ秒）。レコードのフェッチにかかった実際の時間を制御するその他のパラメーター : max_bytes および fetch.min.bytes。デフォルト値は未定義です。このパラメーターは、コンシューマーインスタンスの作成時または REST Proxy の構成ファイルのいずれかで定義された consumer.timeout.ms より小さい場合にのみ使用されます。
  • max_bytes -- 応答に含まれるエンコードされていないキーと値の最大バイト数。これにより、応答のサイズと、デコードされた応答を保存するために必要なメモリーの量を大まかに制御できます。実際の制限は、この設定とサーバー側の構成 consumer.request.max.bytes における最小値です。デフォルトは無制限です。
応答（オブジェクトの JSON 配列）:
 
  • topic (string) -- トピック
  • key (string) -- 組み込みフォーマットに従ってフォーマットされたメッセージキー
  • value (string) -- 組み込みフォーマットに従ってフォーマットされたメッセージ値
  • partition (int) -- メッセージのパーティション
  • offset (long) -- メッセージのオフセット
ステータスコード:
  • 404 Not Found --
    • エラーコード 40403 -- コンシューマーインスタンスが見つかりません
  • 406 Not Acceptable --
    • エラーコード 40601 -- コンシューマーフォーマットが Accept ヘッダーがリクエストした組み込みフォーマットと一致しません。

バイナリリクエストの例:

GET /consumers/testgroup/instances/my_consumer/records?timeout=3000&max_bytes=300000 HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Accept: application/vnd.kafka.binary.v2+json

バイナリ応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.binary.v2+json

[
  {
    "topic": "test",
    "key": "a2V5",
    "value": "Y29uZmx1ZW50",
    "partition": 1,
    "offset": 100,
  },
  {
    "topic": "test",
    "key": "a2V5",
    "value": "a2Fma2E=",
    "partition": 2,
    "offset": 101,
  }
]

Avro リクエストの例 :

GET /consumers/avrogroup/instances/my_avro_consumer/records?timeout=3000&max_bytes=300000 HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Accept: application/vnd.kafka.avro.v2+json

Avro 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.avro.v2+json

[
  {
    "topic": "test",
    "key": 1,
    "value": {
      "id": 1,
      "name": "Bill"
    },
    "partition": 1,
    "offset": 100,
  },
  {
    "topic": "test",
    "key": 2,
    "value": {
      "id": 2,
      "name": "Melinda"
    },
    "partition": 2,
    "offset": 101,
  }
]

JSON リクエストの例 :

GET /consumers/jsongroup/instances/my_json_consumer/records?timeout=3000&max_bytes=300000 HTTP/1.1
Host: proxy-instance.kafkaproxy.example.com
Accept: application/vnd.kafka.json.v2+json

JSON 応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.json.v2+json

[
  {
    "topic": "test",
    "key": "somekey",
    "value": {"foo":"bar"},
    "partition": 1,
    "offset": 10,
  },
  {
    "topic": "test",
    "key": "somekey",
    "value": ["foo", "bar"],
    "partition": 2,
    "offset": 11,
  }
]

ブローカー

ブローカーリソースによって、クラスター内の Kafka ブローカーの現在のステートを確認できます。

GET /brokers

ブローカーのリストを取得します。

レスポンスの JSON オブジェクト:
 
  • brokers (array) -- ブローカー ID のリスト

リクエストの例 :

GET /brokers HTTP/1.1
Host: kafkaproxy.example.com
Accept: application/vnd.kafka.v2+json, application/vnd.kafka+json, application/json

応答の例 :

HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "brokers": [1, 2, 3]
}

REST Proxy API v3

これらの API は、Confluent Server （Confluent Enterprise の一部として）と REST Proxy の両方で使用できます。Confluent Server で API を使用する場合、すべてのパスの先頭に /kafka を付ける必要があります。たとえば、クラスターをリストするパスは以下のようになります。

  • Confluent Server: /kafka/v3/clusters
  • REST Proxy: /v3/clusters

Confluent Server は、v3 の管理用 API 用の Kafka ブローカーで、これらの API の組み込みインスタンスを提供します。組み込み API は、Confluent HTTP サービス confluent.http.server.listeners で実行されます。このため、HTTP サーバーを実行している場合、REST Proxy v3 API は、ブローカーを通じて自動的に利用可能になります。Confluent HTTP サービス上では、追加の構成で使用できる別のエンドポイントとして、Metadata Server（MDS） も実行されています。

ちなみに

以下のセクションも参照してください。

クラスター

GET /clusters

クラスターの一覧表示

既知の Kafka クラスターのリストを返します。現在、Kafka と Kafka REST Proxy はどちらも bootstrap.servers の構成が指す Kafka クラスターしか認識しません。したがって、1 つの Kafka クラスターのみ応答で返されます。

リクエストの例

GET /clusters HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    Kafka クラスターのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaClusterList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaCluster",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1",
                "resource_name": "crn:///kafka=cluster-1"
            },
            "cluster_id": "cluster-1",
            "controller": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1"
            },
            "acls": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/acls"
            },
            "brokers": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers"
            },
            "broker_configs": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/broker-configs"
            },
            "consumer_groups": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups"
            },
            "topics": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics"
            },
            "partition_reassignments": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/-/partitions/-/reassignment"
            }
        }
    ]
}
GET /clusters/{cluster_id}

クラスターの取得

指定された cluster_id を持つ Kafka クラスターを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    Kafka クラスター。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaCluster",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1",
        "resource_name": "crn:///kafka=cluster-1"
    },
    "cluster_id": "cluster-1",
    "controller": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1"
    },
    "acls": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/acls"
    },
    "brokers": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/brokers"
    },
    "broker_configs": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/broker-configs"
    },
    "consumer_groups": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups"
    },
    "topics": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics"
    },
    "partition_reassignments": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/-/partitions/-/reassignment"
    }
}

ACL

GET /clusters/{cluster_id}/acls

ACL の検索

検索基準に一致する ACL のリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
クエリのパラメーター:
  • resource_type (string) -- ACL リソースタイプ。
  • resource_name (string) -- ACL リソース名。
  • pattern_type (string) -- ACL パターンタイプ。
  • principal (string) -- ACL プリンシパル。
  • host (string) -- ACL ホスト。
  • operation (string) -- ACL 操作。
  • permission (string) -- ACL アクセス許可。

リクエストの例

GET /clusters/{cluster_id}/acls HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    ACL のリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaAclList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/acls?principal=alice"
    },
    "data": [
        {
            "kind": "KafkaAcl",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/acls?resource_type=TOPIC&resource_name=topic-&pattern_type=PREFIXED&principal=alice&host=*&operation=ALL&permission=ALLOW"
            },
            "cluster_id": "cluster-1",
            "resource_type": "TOPIC",
            "resource_name": "topic-",
            "pattern_type": "PREFIXED",
            "principal": "alice",
            "host": "*",
            "operation": "ALL",
            "permission": "ALLOW"
        },
        {
            "kind": "KafkaAcl",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/acls?resource_type=CLUSTER&resource_name=cluster-1&pattern_type=LITERAL&principal=bob&host=*&operation=DESCRIBE&permission=DENY"
            },
            "cluster_id": "cluster-1",
            "resource_type": "CLUSTER",
            "resource_name": "cluster-2",
            "pattern_type": "LITERAL",
            "principal": "alice",
            "host": "*",
            "operation": "DESCRIBE",
            "permission": "DENY"
        }
    ]
}
POST /clusters/{cluster_id}/acls

ACL の作成

ACL を作成します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

POST /clusters/{cluster_id}/acls HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "resource_type": "UNKNOWN",
    "resource_name": "string",
    "pattern_type": "UNKNOWN",
    "principal": "string",
    "host": "string",
    "operation": "UNKNOWN",
    "permission": "UNKNOWN"
}
ステータスコード:
DELETE /clusters/{cluster_id}/acls

ACL の削除

検索基準に一致する ACL のリストを削除します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
クエリのパラメーター:
  • resource_type (string) -- ACL リソースタイプ。
  • resource_name (string) -- ACL リソース名。
  • pattern_type (string) -- ACL パターンタイプ。
  • principal (string) -- ACL プリンシパル。
  • host (string) -- ACL ホスト。
  • operation (string) -- ACL 操作。
  • permission (string) -- ACL アクセス許可。
ステータスコード:
  • 200 OK --

    削除された ACL のリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "data": [
        {
            "kind": "KafkaAcl",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/acls?resource_type=TOPIC&resource_name=topic-&pattern_type=PREFIXED&principal=alice&host=*&operation=ALL&permission=ALLOW"
            },
            "cluster_id": "cluster-1",
            "resource_type": "TOPIC",
            "resource_name": "topic-",
            "pattern_type": "PREFIXED",
            "principal": "alice",
            "host": "*",
            "operation": "ALL",
            "permission": "ALLOW"
        },
        {
            "kind": "KafkaAcl",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/acls?resource_type=CLUSTER&resource_name=cluster-1&pattern_type=LITERAL&principal=bob&host=*&operation=DESCRIBE&permission=DENY"
            },
            "cluster_id": "cluster-1",
            "resource_type": "CLUSTER",
            "resource_name": "cluster-2",
            "pattern_type": "LITERAL",
            "principal": "alice",
            "host": "*",
            "operation": "DESCRIBE",
            "permission": "DENY"
        }
    ]
}

構成

GET /clusters/{cluster_id}/broker-configs

クラスター構成の一覧表示

指定された Kafka クラスターの構成パラメーターのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/broker-configs HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    クラスター構成のリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaClusterConfigList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/broker-configs",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaClusterConfig",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/broker-configs/max.connections",
                "resource_name": "crn:///kafka=cluster-1/broker-config=max.connections"
            },
            "cluster_id": "cluster-1",
            "config_type": "BROKER",
            "name": "max.connections",
            "value": "1000",
            "is_default": false,
            "is_read_only": false,
            "is_sensitive": false,
            "source": "DYNAMIC_DEFAULT_BROKER_CONFIG",
            "synonyms": [
                {
                    "name": "max.connections",
                    "value": "1000",
                    "source": "DYNAMIC_DEFAULT_BROKER_CONFIG"
                },
                {
                    "name": "max.connections",
                    "value": "2147483647",
                    "source": "DEFAULT_CONFIG"
                }
            ]
        },
        {
            "kind": "KafkaClusterConfig",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/broker-configs/compression.type",
                "resource_name": "crn:///kafka=cluster-1/broker-config=compression.type"
            },
            "cluster_id": "cluster-1",
            "config_type": "BROKER",
            "name": "compression.type",
            "value": "gzip",
            "is_default": false,
            "is_read_only": false,
            "is_sensitive": false,
            "source": "DYNAMIC_DEFAULT_BROKER_CONFIG",
            "synonyms": [
                {
                    "name": "compression.type",
                    "value": "gzip",
                    "source": "DYNAMIC_DEFAULT_BROKER_CONFIG"
                },
                {
                    "name": "compression.type",
                    "value": "producer",
                    "source": "DEFAULT_CONFIG"
                }
            ]
        }
    ]
}
POST /clusters/{cluster_id}/broker-configs:alter

クラスター構成のバッチ変更

一連の Kafka クラスター構成パラメーターをアップデートまたは削除します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

POST /clusters/{cluster_id}/broker-configs:alter HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "data": [
        {
            "name": "max.connections",
            "operation": "DELETE"
        },
        {
            "name": "compression.type",
            "value": "gzip"
        }
    ]
}
ステータスコード:
GET /clusters/{cluster_id}/broker-configs/{name}

クラスター構成の取得

name で指定された構成パラメーターを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • name (string) -- 構成パラメーター名。

リクエストの例

GET /clusters/{cluster_id}/broker-configs/{name} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    クラスターの構成パラメーター。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaClusterConfig",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/broker-configs/compression.type",
        "resource_name": "crn:///kafka=cluster-1/broker-config=compression.type"
    },
    "cluster_id": "cluster-1",
    "config_type": "BROKER",
    "name": "compression.type",
    "value": "gzip",
    "is_default": false,
    "is_read_only": false,
    "is_sensitive": false,
    "source": "DYNAMIC_DEFAULT_BROKER_CONFIG",
    "synonyms": [
        {
            "name": "compression.type",
            "value": "gzip",
            "source": "DYNAMIC_DEFAULT_BROKER_CONFIG"
        },
        {
            "name": "compression.type",
            "value": "producer",
            "source": "DEFAULT_CONFIG"
        }
    ]
}
PUT /clusters/{cluster_id}/broker-configs/{name}

クラスター構成のアップデート

name で指定された構成パラメーターをアップデートします。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • name (string) -- 構成パラメーター名。

リクエストの例

PUT /clusters/{cluster_id}/broker-configs/{name} HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "value": "gzip"
}
ステータスコード:
DELETE /clusters/{cluster_id}/broker-configs/{name}

クラスター構成のリセット

name で指定された構成パラメーターをデフォルト値にリセットします。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • name (string) -- 構成パラメーター名。
ステータスコード:
GET /clusters/{cluster_id}/brokers/{broker_id}/configs

ブローカー構成の一覧表示

指定された Kafka ブローカーに属する構成パラメーターのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。

リクエストの例

GET /clusters/{cluster_id}/brokers/{broker_id}/configs HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    ブローカー構成のリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaBrokerConfigList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/brokers/1/configs",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaBrokerConfig",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/brokers/1/configs/max.connections",
                "resource_name": "crn:///kafka=cluster-1/broker=1/config=max.connections"
            },
            "cluster_id": "cluster-1",
            "broker_id": 1,
            "name": "max.connections",
            "value": "1000",
            "is_default": false,
            "is_read_only": false,
            "is_sensitive": false,
            "source": "DYNAMIC_BROKER_CONFIG",
            "synonyms": [
                {
                    "name": "max.connections",
                    "value": "1000",
                    "source": "DYNAMIC_BROKER_CONFIG"
                },
                {
                    "name": "max.connections",
                    "value": "2147483647",
                    "source": "DEFAULT_CONFIG"
                }
            ]
        },
        {
            "kind": "KafkaBrokerConfig",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/brokers/1/configs/compression.type",
                "resource_name": "crn:///kafka=cluster-1/broker=1/config=compression.type"
            },
            "cluster_id": "cluster-1",
            "broker_id": 1,
            "name": "compression.type",
            "value": "gzip",
            "is_default": false,
            "is_read_only": false,
            "is_sensitive": false,
            "source": "DYNAMIC_BROKER_CONFIG",
            "synonyms": [
                {
                    "name": "compression.type",
                    "value": "gzip",
                    "source": "DYNAMIC_BROKER_CONFIG"
                },
                {
                    "name": "compression.type",
                    "value": "producer",
                    "source": "DEFAULT_CONFIG"
                }
            ]
        }
    ]
}
POST /clusters/{cluster_id}/brokers/{broker_id}/configs:alter

ブローカー構成のバッチ変更

一連のブローカー構成パラメーターをアップデートまたは削除します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。

リクエストの例

POST /clusters/{cluster_id}/brokers/{broker_id}/configs:alter HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "data": [
        {
            "name": "max.connections",
            "operation": "DELETE"
        },
        {
            "name": "compression.type",
            "value": "gzip"
        }
    ]
}
ステータスコード:
GET /clusters/{cluster_id}/brokers/{broker_id}/configs/{name}

ブローカー構成の取得

name で指定された構成パラメーターを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。
  • name (string) -- 構成パラメーター名。

リクエストの例

GET /clusters/{cluster_id}/brokers/{broker_id}/configs/{name} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    ブローカーの構成パラメーター。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaBrokerConfig",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/brokers/1/configs/compression.type",
        "resource_name": "crn:///kafka=cluster-1/broker=1/config=compression.type"
    },
    "cluster_id": "cluster-1",
    "broker_id": 1,
    "name": "compression.type",
    "value": "gzip",
    "is_default": false,
    "is_read_only": false,
    "is_sensitive": false,
    "source": "DYNAMIC_BROKER_CONFIG",
    "synonyms": [
        {
            "name": "compression.type",
            "value": "gzip",
            "source": "DYNAMIC_BROKER_CONFIG"
        },
        {
            "name": "compression.type",
            "value": "producer",
            "source": "DEFAULT_CONFIG"
        }
    ]
}
PUT /clusters/{cluster_id}/brokers/{broker_id}/configs/{name}

ブローカー構成のアップデート

name で指定された構成パラメーターをアップデートします。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。
  • name (string) -- 構成パラメーター名。

リクエストの例

PUT /clusters/{cluster_id}/brokers/{broker_id}/configs/{name} HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "value": "gzip"
}
ステータスコード:
DELETE /clusters/{cluster_id}/brokers/{broker_id}/configs/{name}

ブローカー構成のリセット

name で指定された構成パラメーターをデフォルト値にリセットします。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。
  • name (string) -- 構成パラメーター名。
ステータスコード:
GET /clusters/{cluster_id}/topics/{topic_name}/configs

トピック構成の一覧表示

指定されたトピックに属する構成のリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。

リクエストの例

GET /clusters/{cluster_id}/topics/{topic_name}/configs HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    クラスター構成のリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaTopicConfigList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/configs",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaTopicConfig",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/configs/cleanup.policy",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/config=cleanup.policy"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "name": "cleanup.policy",
            "value": "compact",
            "is_default": false,
            "is_read_only": false,
            "is_sensitive": false,
            "source": "DYNAMIC_TOPIC_CONFIG",
            "synonyms": [
                {
                    "name": "cleanup.policy",
                    "value": "compact",
                    "source": "DYNAMIC_TOPIC_CONFIG"
                },
                {
                    "name": "cleanup.policy",
                    "value": "delete",
                    "source": "DEFAULT_CONFIG"
                }
            ]
        },
        {
            "kind": "KafkaTopicConfig",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/configs/compression.type",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/config=compression.type"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "name": "compression.type",
            "value": "gzip",
            "is_default": false,
            "is_read_only": false,
            "is_sensitive": false,
            "source": "DYNAMIC_TOPIC_CONFIG",
            "synonyms": [
                {
                    "name": "compression.type",
                    "value": "gzip",
                    "source": "DYNAMIC_TOPIC_CONFIG"
                },
                {
                    "name": "compression.type",
                    "value": "producer",
                    "source": "DEFAULT_CONFIG"
                }
            ]
        }
    ]
}
POST /clusters/{cluster_id}/topics/{topic_name}/configs:alter

トピック構成のバッチ変更

一連のトピック構成をアップデートまたは削除します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。

リクエストの例

POST /clusters/{cluster_id}/topics/{topic_name}/configs:alter HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "data": [
        {
            "name": "cleanup.policy",
            "operation": "DELETE"
        },
        {
            "name": "compression.type",
            "value": "gzip"
        }
    ]
}
ステータスコード:
GET /clusters/{cluster_id}/topics/{topic_name}/configs/{name}

トピック構成の取得

指定した "name" の構成を返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。
  • name (string) -- 構成パラメーター名。

リクエストの例

GET /clusters/{cluster_id}/topics/{topic_name}/configs/{name} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    トピック構成パラメーター。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaTopicConfig",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/compression.type",
        "resource_name": "crn:///kafka=cluster-1/topic=topic-1/config=compression.type"
    },
    "cluster_id": "cluster-1",
    "topic_name": "topic-1",
    "name": "compression.type",
    "value": "gzip",
    "is_default": false,
    "is_read_only": false,
    "is_sensitive": false,
    "source": "DYNAMIC_TOPIC_CONFIG",
    "synonyms": [
        {
            "name": "compression.type",
            "value": "gzip",
            "source": "DYNAMIC_TOPIC_CONFIG"
        },
        {
            "name": "compression.type",
            "value": "producer",
            "source": "DEFAULT_CONFIG"
        }
    ]
}
PUT /clusters/{cluster_id}/topics/{topic_name}/configs/{name}

トピック構成のアップデート

指定した "name" の構成をアップデートします。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。
  • name (string) -- 構成パラメーター名。

リクエストの例

PUT /clusters/{cluster_id}/topics/{topic_name}/configs/{name} HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "value": "gzip"
}
ステータスコード:
DELETE /clusters/{cluster_id}/topics/{topic_name}/configs/{name}

トピック構成のリセット

指定した "name" の構成をデフォルト値にリセットします。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。
  • name (string) -- 構成パラメーター名。
ステータスコード:

ブローカー

GET /clusters/{cluster_id}/brokers

ブローカーの一覧表示

指定された Kafka クラスターに属するブローカーのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/brokers HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    ブローカーのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaBrokerList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/brokers",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaBroker",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/brokers/1",
                "resource_name": "crn:///kafka=cluster-1/broker=1"
            },
            "cluster_id": "cluster-1",
            "broker_id": 1,
            "host": "localhost",
            "port": 9291,
            "configs": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1/configs"
            },
            "partition_replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1/partition-replicas"
            }
        },
        {
            "kind": "KafkaBroker",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/brokers/2",
                "resource_name": "crn:///kafka=cluster-1/broker=2"
            },
            "cluster_id": "cluster-1",
            "broker_id": 2,
            "host": "localhost",
            "port": 9292,
            "configs": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/2/configs"
            },
            "partition_replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/2/partition-replicas"
            }
        },
        {
            "kind": "KafkaBroker",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/brokers/3",
                "resource_name": "crn:///kafka=cluster-1/broker=3"
            },
            "cluster_id": "cluster-1",
            "broker_id": 3,
            "host": "localhost",
            "port": 9293,
            "configs": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/3/configs"
            },
            "partition_replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/3/partition-replicas"
            }
        }
    ]
}
GET /clusters/{cluster_id}/brokers/{broker_id}

ブローカーの取得

broker_id で指定されたブローカーを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。

リクエストの例

GET /clusters/{cluster_id}/brokers/{broker_id} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    ブローカー。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaBroker",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/brokers/1",
        "resource_name": "crn:///kafka=cluster-1/broker=1"
    },
    "cluster_id": "cluster-1",
    "broker_id": 1,
    "host": "localhost",
    "port": 9291,
    "configs": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1/configs"
    },
    "partition_replicas": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1/partition-replicas"
    }
}
DELETE /clusters/{cluster_id}/brokers/{broker_id}

ブローカーの削除

broker_id で指定されたブローカーを削除します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。
ステータスコード:
  • 202 Accepted -- 受け入れ済み
  • 400 Bad Request --

    無効なリクエスト

    IllegalBrokerRemoval:

    HTTP/1.1 400 Bad Request
Content-Type: application/json

{
    "error_code": 400,
    "message": "Cannot remove broker 1 as there are partitions with replication factor equal to 1 on the broker. One such partition: test_topic_parition_0."
}

    BalancerOffline:

    HTTP/1.1 400 Bad Request
Content-Type: application/json

{
    "error_code": 400,
    "message": "The Confluent Balancer component is disabled or not started yet."
}
  • 404 Not Found --

    ブローカーが見つかりません。

    応答の例:

    HTTP/1.1 404 Not Found
Content-Type: application/json

{
    "error_code": 404,
    "message": "Broker not found. Broker: 1 not found in the cluster: cluster-1"
}
GET /clusters/{cluster_id}/brokers/{broker_id}/partition-replicas

ブローカーごとのレプリカの検索

指定されたブローカーに割り当てられたレプリカのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。

リクエストの例

GET /clusters/{cluster_id}/brokers/{broker_id}/partition-replicas HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    レプリカのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaReplicaList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/brokers/1/partition-replicas",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaReplica",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/2/replicas/1",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=2/replica=1"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 2,
            "broker_id": 1,
            "is_leader": true,
            "is_in_sync": true,
            "broker": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1"
            }
        },
        {
            "kind": "KafkaReplica",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-2/partitions/3/replicas/1",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-3/partition=3/replica=1"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-2",
            "partition_id": 3,
            "broker_id": 1,
            "is_leader": false,
            "is_in_sync": true,
            "broker": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1"
            }
        },
        {
            "kind": "KafkaReplica",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-3/partitions/1/replicas/1",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-3/partition=1/replica=1"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-3",
            "partition_id": 1,
            "broker_id": 1,
            "is_leader": false,
            "is_in_sync": false,
            "broker": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1"
            }
        }
    ]
}

コンシューマーグループ

GET /clusters/{cluster_id}/consumer-groups

コンシューマーグループの一覧表示

指定された Kafka クラスターに属するコンシューマーグループのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/consumer-groups HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    コンシューマーグループのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaConsumerGroupList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaConsumerGroup",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "is_simple": false,
            "partition_assignor": "org.apache.kafka.clients.consumer.RoundRobinAssignor",
            "state": "STABLE",
            "coordinator": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1"
            },
            "consumers": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers"
            },
            "lag_summary": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/lag-summary"
            }
        },
        {
            "kind": "KafkaConsumerGroup",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-2",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-2"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-2",
            "is_simple": false,
            "partition_assignor": "org.apache.kafka.clients.consumer.StickyAssignor",
            "state": "PREPARING_REBALANCE",
            "coordinator": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/2"
            },
            "consumers": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-2/consumers"
            },
            "lag_summary": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-2/lag-summary"
            }
        },
        {
            "kind": "KafkaConsumerGroup",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-3",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-3"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-3",
            "is_simple": false,
            "partition_assignor": "org.apache.kafka.clients.consumer.RangeAssignor",
            "state": "DEAD",
            "coordinator": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/3"
            },
            "consumers": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-3/consumers"
            },
            "lag_summary": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-3/lag-summary"
            }
        }
    ]
}
GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}

コンシューマーグループの取得

consumer_group_id で指定されたコンシューマーグループを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • consumer_group_id (string) -- コンシューマーグループ ID。

リクエストの例

GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    コンシューマーグループ。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaConsumerGroup",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1",
        "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1"
    },
    "cluster_id": "cluster-1",
    "consumer_group_id": "consumer-group-1",
    "is_simple": false,
    "partition_assignor": "org.apache.kafka.clients.consumer.RoundRobinAssignor",
    "state": "STABLE",
    "coordinator": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1"
    },
    "consumers": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers"
    },
    "lag_summary": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/lag-summary"
    }
}
GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/consumers

コンシューマーの一覧表示

指定されたコンシューマーグループに属するコンシューマーのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • consumer_group_id (string) -- コンシューマーグループ ID。

リクエストの例

GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/consumers HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    コンシューマーのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaConsumerList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaConsumer",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/consumer=consumer-1"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "consumer_id": "consumer-1",
            "instance_id": "consumer-instance-1",
            "client_id": "client-1",
            "assignments": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1/assignments"
            }
        },
        {
            "kind": "KafkaConsumer",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-2",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/consumer=consumer-2"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "consumer_id": "consumer-2",
            "instance_id": "consumer-instance-2",
            "client_id": "client-2",
            "assignments": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-2/assignments"
            }
        },
        {
            "kind": "KafkaConsumer",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-2",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/consumer=consumer-2"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "consumer_id": "consumer-2",
            "instance_id": "consumer-instance-2",
            "client_id": "client-2",
            "assignments": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-2/assignments"
            }
        }
    ]
}
GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/lag-summary

コンシューマーグループラグの概要の取得

指定されたコンシューマーグループに属するコンシューマーの最大ラグと合計ラグを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • consumer_group_id (string) -- コンシューマーグループ ID。

リクエストの例

GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/lag-summary HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    単一コンシューマーグループ内の最大および合計コンシューマーラグ。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaConsumerGroupLagSummary",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/lag-summary",
        "resource_name": "crn:///kafka=cluster-1/consumer-groups=consumer-group-1/lag-summary"
    },
    "cluster_id": "cluster-1",
    "consumer_group_id": "consumer-group-1",
    "max_lag_consumer_id": "consumer-1",
    "max_lag_instance_id": "consumer-instance-1",
    "max_lag_client_id": "client-1",
    "max_lag_topic_name": "topic-1",
    "max_lag_partition_id": 1,
    "max_lag": 100,
    "total_lag": 110,
    "max_lag_consumer": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1"
    },
    "max_lag_partition": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1"
    }
}
GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/lags

コンシューマーラグの一覧表示

指定されたコンシューマーグループに属するコンシューマーについて、コンシューマーラグのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • consumer_group_id (string) -- コンシューマーグループ ID。

リクエストの例

GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/lags HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    コンシューマーラグのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaConsumerLagList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/lags",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaConsumerLag",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/lags/topic-1/partitions/1",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/lag=topic-1/partition=1"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "topic_name": "topic-1",
            "partition_id": 1,
            "consumer_id": "consumer-1",
            "instance_id": "consumer-instance-1",
            "client_id": "client-1",
            "current_offset": 1,
            "log_end_offset": 101,
            "lag": 100
        },
        {
            "kind": "KafkaConsumerLag",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/lags/topic-1/partitions/2",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/lag=topic-1/partition=2"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "topic_name": "topic-1",
            "partition_id": 2,
            "consumer_id": "consumer-2",
            "instance_id": "consumer-instance-2",
            "client_id": "client-2",
            "current_offset": 1,
            "log_end_offset": 11,
            "lag": 10
        },
        {
            "kind": "KafkaConsumerLag",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/lags/topic-1/partitions/3",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/lag=topic-1/partition=3"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "topic_name": "topic-1",
            "partition_id": 3,
            "consumer_id": "consumer-3",
            "instance_id": "consumer-instance-3",
            "client_id": "client-3",
            "current_offset": 1,
            "log_end_offset": 1,
            "lag": 0
        }
    ]
}
GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/consumers/{consumer_id}

コンシューマーの取得

consumer_id で指定されたコンシューマーを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • consumer_group_id (string) -- コンシューマーグループ ID。
  • consumer_id (string) -- コンシューマー ID。

リクエストの例

GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/consumers/{consumer_id} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    コンシューマー。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaConsumer",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1",
        "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/consumer=consumer-1"
    },
    "cluster_id": "cluster-1",
    "consumer_group_id": "consumer-group-1",
    "consumer_id": "consumer-1",
    "instance_id": "consumer-instance-1",
    "client_id": "client-1",
    "assignments": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1/assignments"
    }
}
GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/consumers/{consumer_id}/assignments

コンシューマー割り当ての一覧表示

指定されたコンシューマーのパーティション割り当てのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • consumer_group_id (string) -- コンシューマーグループ ID。
  • consumer_id (string) -- コンシューマー ID。

リクエストの例

GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/consumers/{consumer_id}/assignments HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    コンシューマーグループ割り当てのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaConsumerAssignmentList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1/assignments",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaConsumerAssignment",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1/assignments/topic-1/partitions/1",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/consumer=consumer-1/assignment=topic=1/partition=1"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "consumer_id": "consumer-1",
            "topic_name": "topic-1",
            "partition_id": 1,
            "partition": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1"
            },
            "lag": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/lags/consumer-group-1"
            }
        },
        {
            "kind": "KafkaConsumerAssignment",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1/assignments/topic-2/partitions/2",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/consumer=consumer-1/assignment=topic=2/partition=2"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "consumer_id": "consumer-1",
            "topic_name": "topic-2",
            "partition_id": 2,
            "partition": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-2/partitions/2"
            },
            "lag": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-2/partitions/2/lags/consumer-groups-1"
            }
        },
        {
            "kind": "KafkaConsumerAssignment",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1/assignments/topic-3/partitions/3",
                "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/consumer=consumer-1/assignment=topic=3/partition=3"
            },
            "cluster_id": "cluster-1",
            "consumer_group_id": "consumer-group-1",
            "consumer_id": "consumer-1",
            "topic_name": "topic-3",
            "partition_id": 3,
            "partition": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-3/partitions/3"
            },
            "lag": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-3/partitions/3/lags/consumer-groups-1"
            }
        }
    ]
}
GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/consumers/{consumer_id}/assignments/{topic_name}/partitions/{partition_id}

コンシューマー割り当ての取得

指定されたパーティションに対して指定したコンシューマー割り当てに関する情報を返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • consumer_group_id (string) -- コンシューマーグループ ID。
  • consumer_id (string) -- コンシューマー ID。
  • topic_name (string) -- トピック名。
  • partition_id (integer) -- パーティション ID。

リクエストの例

GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/consumers/{consumer_id}/assignments/{topic_name}/partitions/{partition_id} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    コンシューマーグループの割り当て。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaConsumerAssignment",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/consumers/consumer-1/assignments/topic-1/partitions/1",
        "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/consumer=consumer-1/assignment=topic=1/partition=1"
    },
    "cluster_id": "cluster-1",
    "consumer_group_id": "consumer-group-1",
    "consumer_id": "consumer-1",
    "topic_name": "topic-1",
    "partition_id": 1,
    "partition": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1"
    },
    "lag": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/lags/consumer-group-1"
    }
}

パーティション

GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/lags/{topic_name}/partitions/{partition_id}

コンシューマーラグの取得

指定した "partition_id" を持つパーティション上のコンシューマーラグを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • consumer_group_id (string) -- コンシューマーグループ ID。
  • topic_name (string) -- トピック名。
  • partition_id (integer) -- パーティション ID。

リクエストの例

GET /clusters/{cluster_id}/consumer-groups/{consumer_group_id}/lags/{topic_name}/partitions/{partition_id} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    コンシューマーラグ。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaConsumerLag",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/consumer-groups/consumer-group-1/lags/topic-1/partitions/1",
        "resource_name": "crn:///kafka=cluster-1/consumer-group=consumer-group-1/lag=topic-1/partition=1"
    },
    "cluster_id": "cluster-1",
    "consumer_group_id": "consumer-group-1",
    "topic_name": "topic-1",
    "partition_id": 1,
    "consumer_id": "consumer-1",
    "instance_id": "consumer-instance-1",
    "client_id": "client-1",
    "current_offset": 1,
    "log_end_offset": 101,
    "lag": 100
}
GET /clusters/{cluster_id}/topics/{topic_name}/partitions

パーティションの一覧表示

指定されたトピックに属するパーティションのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。

リクエストの例

GET /clusters/{cluster_id}/topics/{topic_name}/partitions HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    パーティションのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaPartitionList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaPartition",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=1"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 1,
            "leader": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas/1"
            },
            "replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas"
            },
            "reassignment": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/reassignment"
            }
        },
        {
            "kind": "KafkaPartition",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/2",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=2"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 2,
            "leader": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/2/replicas/2"
            },
            "replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/2/replicas"
            },
            "reassignment": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/2/reassignment"
            }
        },
        {
            "kind": "KafkaPartition",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/3",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=3"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 3,
            "leader": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/3/replicas/3"
            },
            "replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/3/replicas"
            },
            "reassignment": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/3/reassignment"
            }
        }
    ]
}
GET /clusters/{cluster_id}/topics/{topic_name}/partitions/{partition_id}

パーティションの取得

指定した "partition_id" を持つパーティションを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。
  • partition_id (integer) -- パーティション ID。

リクエストの例

GET /clusters/{cluster_id}/topics/{topic_name}/partitions/{partition_id} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    パーティション

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaPartition",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1",
        "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=1"
    },
    "cluster_id": "cluster-1",
    "topic_name": "topic-1",
    "partition_id": 1,
    "leader": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas/1"
    },
    "replicas": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas"
    },
    "reassignment": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/reassignment"
    }
}
GET /clusters/{cluster_id}/topics/-/partitions/-/reassignment

すべてのレプリカの再割り当ての一覧表示

指定した Kafka クラスターで実行中のすべてのレプリカの再割り当てのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/topics/-/partitions/-/reassignment HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    実行中のレプリカの再割り当て。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaReassignmentList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/-/partitions/-/reassignment",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaReassignment",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/reassignment",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=1/reassignment"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 1,
            "adding_replicas": [
                1,
                2
            ],
            "removing_replicas": [
                3
            ],
            "replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas"
            }
        },
        {
            "kind": "KafkaReassignment",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-2/partitions/2/reassignment",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-2/partition=2/reassignment"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-2",
            "partition_id": 2,
            "adding_replicas": [
                1
            ],
            "removing_replicas": [
                2,
                3
            ],
            "replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-2/partitions/2/replicas"
            }
        },
        {
            "kind": "KafkaReassignment",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-3/partitions/3/reassignment",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-3/partition=3/reassignment"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-3",
            "partition_id": 3,
            "adding_replicas": [
                3
            ],
            "removing_replicas": [
                1,
                2
            ],
            "replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-3/partitions/3/replicas"
            }
        }
    ]
}
GET /clusters/{cluster_id}/topics/{topic_name}/partitions/-/reassignment

トピックごとのレプリカの再割り当ての検索

指定したトピックについて実行中のレプリカの再割り当てのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。

リクエストの例

GET /clusters/{cluster_id}/topics/{topic_name}/partitions/-/reassignment HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    実行中のレプリカの再割り当て。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaReassignmentList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/-/partitions/-/reassignment",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaReassignment",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/reassignment",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=1/reassignment"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 1,
            "adding_replicas": [
                1,
                2
            ],
            "removing_replicas": [
                3
            ],
            "replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas"
            }
        },
        {
            "kind": "KafkaReassignment",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/2/reassignment",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=2/reassignment"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 2,
            "adding_replicas": [
                1
            ],
            "removing_replicas": [
                2,
                3
            ],
            "replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/2/replicas"
            }
        },
        {
            "kind": "KafkaReassignment",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/3/reassignment",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=3/reassignment"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 3,
            "adding_replicas": [
                3
            ],
            "removing_replicas": [
                1,
                2
            ],
            "replicas": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/3/replicas"
            }
        }
    ]
}
GET /clusters/{cluster_id}/topics/{topic_name}/partitions/{partition_id}/reassignment

レプリカの再割り当ての取得

指定したパーティションについて実行中のレプリカの再割り当てのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。
  • partition_id (integer) -- パーティション ID。

リクエストの例

GET /clusters/{cluster_id}/topics/{topic_name}/partitions/{partition_id}/reassignment HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    実行中のレプリカの再割り当て。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaReassignment",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/reassignment",
        "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=1/reassignment"
    },
    "cluster_id": "cluster-1",
    "topic_name": "topic-1",
    "partition_id": 1,
    "adding_replicas": [
        1,
        2
    ],
    "removing_replicas": [
        3
    ],
    "replicas": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas"
    }
}

Topic

GET /clusters/{cluster_id}/topics

トピックの一覧表示

指定された Kafka クラスターに属するトピックのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/topics HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    トピックのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaTopicList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaTopic",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "is_internal": false,
            "replication_factor": 3,
            "partitions": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions"
            },
            "configs": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/configs"
            },
            "partition_reassignments": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/-/reassignments"
            }
        },
        {
            "kind": "KafkaTopic",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-2",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-2"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-2",
            "is_internal": true,
            "replication_factor": 4,
            "partitions": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-2/partitions"
            },
            "configs": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-2/configs"
            },
            "partition_reassignments": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-2/partitions/-/reassignments"
            }
        },
        {
            "kind": "KafkaTopic",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-3",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-3"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-3",
            "is_internal": false,
            "replication_factor": 5,
            "partitions": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-3/partitions"
            },
            "configs": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-3/configs"
            },
            "partition_reassignments": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-3/partitions/-/reassignments"
            }
        }
    ]
}
POST /clusters/{cluster_id}/topics

トピックの作成

トピックを作成します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

uniform_replication:

POST /clusters/{cluster_id}/topics HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "topic_name": "topic-X",
    "partitions_count": 64,
    "replication_factor": 3,
    "configs": [
        {
            "name": "cleanup.policy",
            "value": "compact"
        },
        {
            "name": "compression.type",
            "value": "gzip"
        }
    ]
}

explicit_replicas_assignments:

POST /clusters/{cluster_id}/topics HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "topic_name": "topic-Y",
    "replicas_assignments": [
        {
            "partition_id": 0,
            "broker_ids": [
                1,
                2
            ]
        },
        {
            "partition_id": 1,
            "broker_ids": [
                2,
                3
            ]
        },
        {
            "partition_id": 2,
            "broker_ids": [
                3,
                1
            ]
        }
    ],
    "configs": [
        {
            "name": "cleanup.policy",
            "value": "compact"
        },
        {
            "name": "compression.type",
            "value": "gzip"
        }
    ]
}
ステータスコード:
  • 201 Created --

    作成されたトピック。

    応答の例:

    HTTP/1.1 201 Created
Content-Type: application/json

{
    "kind": "KafkaTopic",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-X",
        "resource_name": "crn:///kafka=cluster-1/topic=topic-X"
    },
    "cluster_id": "cluster-1",
    "topic_name": "topic-X",
    "is_internal": false,
    "replication_factor": 3,
    "partitions": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-X/partitions"
    },
    "configs": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-X/configs"
    },
    "partition_reassignments": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-X/partitions/-/reassignments"
    }
}
GET /clusters/{cluster_id}/topics/{topic_name}

トピックの取得

指定した "topic_name" を持つトピックを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。

リクエストの例

GET /clusters/{cluster_id}/topics/{topic_name} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    トピック。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaTopic",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1",
        "resource_name": "crn:///kafka=cluster-1/topic=topic-1"
    },
    "cluster_id": "cluster-1",
    "topic_name": "topic-1",
    "is_internal": false,
    "replication_factor": 3,
    "partitions": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions"
    },
    "configs": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/configs"
    },
    "partition_reassignments": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/-/reassignments"
    }
}
DELETE /clusters/{cluster_id}/topics/{topic_name}

トピックの削除

指定した topic_name を持つトピックを削除します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。
ステータスコード:

レプリカ

GET /clusters/{cluster_id}/topics/{topic_name}/partitions/{partition_id}/replicas

レプリカの一覧表示

指定されたパーティションについてレプリカのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。
  • partition_id (integer) -- パーティション ID。

リクエストの例

GET /clusters/{cluster_id}/topics/{topic_name}/partitions/{partition_id}/replicas HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    レプリカのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaReplicaList",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaReplica",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas/1",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=1/replica=1"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 1,
            "broker_id": 1,
            "is_leader": true,
            "is_in_sync": true,
            "broker": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1"
            }
        },
        {
            "kind": "KafkaReplica",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas/2",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=1/replica=2"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 1,
            "broker_id": 2,
            "is_leader": false,
            "is_in_sync": true,
            "broker": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/2"
            }
        },
        {
            "kind": "KafkaReplica",
            "metadata": {
                "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas/3",
                "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=1/replica=3"
            },
            "cluster_id": "cluster-1",
            "topic_name": "topic-1",
            "partition_id": 1,
            "broker_id": 3,
            "is_leader": false,
            "is_in_sync": false,
            "broker": {
                "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/3"
            }
        }
    ]
}
GET /clusters/{cluster_id}/topics/{topic_name}/partitions/{partition_id}/replicas/{broker_id}

レプリカの取得

指定されたブローカーに割り当てられた、指定されたパーティションのレプリカを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。
  • partition_id (integer) -- パーティション ID。
  • broker_id (integer) -- Kafka ブローカー ID。

リクエストの例

GET /clusters/{cluster_id}/topics/{topic_name}/partitions/{partition_id}/replicas/{broker_id} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    レプリカ。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaReplica",
    "metadata": {
        "self": "http://localhost:9391/v3/clusters/cluster-1/topics/topic-1/partitions/1/replicas/1",
        "resource_name": "crn:///kafka=cluster-1/topic=topic-1/partition=1/replica=1"
    },
    "cluster_id": "cluster-1",
    "topic_name": "topic-1",
    "partition_id": 1,
    "broker_id": 1,
    "is_leader": true,
    "is_in_sync": true,
    "broker": {
        "related": "http://localhost:9391/v3/clusters/cluster-1/brokers/1"
    }
}

レコード（v3）

POST /clusters/{cluster_id}/topics/{topic_name}/records

指定したトピックへのレコードの生成

[![Early Access](https://img.shields.io/badge/Lifecycle%20Stage-Early%20Access-%2345c6e8)](#section/Versioning/API-Lifecycle-Policy)

指定したトピックに対してレコードを生成し、生成された各レコードについてデリバリーレポートを返します。"Transfer-Encoding: chunked" ヘッダーを設定すると、この API をストリーミングモードで使用することができます。接続が開いたままである限り、サーバーはレコードを受け入れ続けます。サーバーに送信された各レコードについて、サーバーはデリバリーレポートを同じ順序で非同期に送り返します。レコードは、サーバーとの間で連結 JSON としてストリーミングされます。エラーはレコードごとに報告されます。接続が正常に確立されている限り、HTTP ステータスコードは HTTP 200 OK になります。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • topic_name (string) -- トピック名。

binary_and_avro_with_subject_and_raw_schema:

POST /clusters/{cluster_id}/topics/{topic_name}/records HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "partition_id": 1,
    "headers": [
        {
            "name": "Header-1",
            "value": "SGVhZGVyLTE="
        },
        {
            "name": "Header-2",
            "value": "SGVhZGVyLTI="
        }
    ],
    "key": {
        "type": "BINARY",
        "data": "Zm9vYmFy"
    },
    "value": {
        "type": "AVRO",
        "subject": "topic-1-key",
        "schema": "{\\\"type\\\":\\\"string\\\"}",
        "data": "foobar"
    },
    "timestamp": "2021-02-05T19:14:42Z"
}

schema_id_and_schema_version:

POST /clusters/{cluster_id}/topics/{topic_name}/records HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "key": {
        "subject_name_strategy": "TOPIC_NAME",
        "schema_id": 1,
        "data": 1000
    },
    "value": {
        "schema_version": 1,
        "data": {
            "foo": "bar"
        }
    }
}

latest_schema:

POST /clusters/{cluster_id}/topics/{topic_name}/records HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "key": {
        "data": 1000
    },
    "value": {
        "data": "foobar"
    }
}

null_and_empty_data:

POST /clusters/{cluster_id}/topics/{topic_name}/records HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "key": {
        "schema_id": 1
    },
    "value": {
        "schema_version": 1,
        "data": null
    }
}

empty_value:

POST /clusters/{cluster_id}/topics/{topic_name}/records HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "key": {
        "data": 1000
    }
}
ステータスコード:
  • 200 OK --

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "cluster_id": "cluster-1",
    "topic_name": "topic-1",
    "partition_id": 1,
    "offset": 0,
    "timestamp": "2021-02-05T19:14:42Z",
    "key": {
        "type": "BINARY",
        "size": 7
    },
    "value": {
        "type": "AVRO",
        "subject": "topic-1-value",
        "schema_id": 1,
        "schema_version": 1,
        "size": 7
    }
}

バランサーステータス

GET /clusters/{cluster_id}/balancer

バランサーのステータスの取得

cluster_id で指定されたクラスターのバランサーコンポーネントに関するステータスを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/balancer HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    バランサーステータス

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaBalancerStatus",
    "metadata": {
        "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/balancer",
        "resource_name": "crn:///kafka=cluster-1/balancer"
    },
    "cluster_id": "cluster-1",
    "status": "ERROR",
    "error_code": 10014,
    "error_message": "The Confluent Balancer failed to start as JBOD is enabled for the cluster.",
    "any_uneven_load": {
        "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/even-cluster-load"
    },
    "broker_tasks": {
        "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/-/tasks"
    }
}
GET /clusters/{cluster_id}/balancer/any-uneven-load

AnyUnevenLoad ステータスの取得

cluster_id で指定されたクラスターの AnyUnevenLoad に関するステータスを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/balancer/any-uneven-load HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    AnyUnevenLoad ステータス

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaAnyUnevenLoad",
    "metadata": {
        "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/any-uneven-load",
        "resource_name": "crn:///kafka=cluster-1/any-uneven-load"
    },
    "cluster_id": "cluster-1",
    "status": "BALANCING",
    "previous_status": "BALANCING_FAILED",
    "status_updated_at": "2019-10-12T07:20:50Z",
    "previous_status_updated_at": "2019-10-12T07:20:35Z",
    "error_code": 10013,
    "error_message": "The Confluent Balancer operation was overridden by a higher priority operation.",
    "broker_tasks": {
        "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/-/tasks"
    }
}
  • 400 Bad Request --

    バランサーオフライン

    応答の例:

    HTTP/1.1 400 Bad Request
Content-Type: application/json

{
    "error_code": 400,
    "message": "The Confluent Balancer component is disabled or not started yet."
}

ブローカータスク

GET /clusters/{cluster_id}/brokers/-/tasks

ブローカータスクの一覧表示

cluster_id で指定されたクラスターの、すべてのブローカーの全タスクのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/brokers/-/tasks HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    タスクのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaBrokerTaskList",
    "metadata": {
        "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/-/tasks",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaBrokerTask",
            "metadata": {
                "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1/tasks/add-broker",
                "resource_name": "crn:///kafka=cluster-1/broker=1/task=add-broker"
            },
            "cluster_id": "cluster_id",
            "broker_id": 1,
            "task_type": "add-broker",
            "task_status": "SUCCESS",
            "sub_task_statuses": {
                "partition_reassignment_status": "COMPLETED"
            },
            "created_at": "2019-10-12T10:20:40Z",
            "updated_at": "2019-10-12T10:20:45Z",
            "broker": {
                "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1"
            }
        },
        {
            "kind": "KafkaBrokerTask",
            "metadata": {
                "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/2/tasks/remove-broker",
                "resource_name": "crn:///kafka=cluster-1/broker=2/task=remove-broker"
            },
            "cluster_id": "cluster_id",
            "broker_id": 2,
            "task_type": "remove-broker",
            "task_status": "FAILED",
            "sub_task_statuses": {
                "partition_reassignment_status": "ERROR",
                "broker_shutdown_status": "CANCELED"
            },
            "created_at": "2019-10-12T07:20:50Z",
            "updated_at": "2019-10-12T07:20:55Z",
            "error_code": 10006,
            "error_message": "Error while computing the initial remove broker plan for brokers [2] prior to shutdown.",
            "broker": {
                "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/2"
            }
        }
    ]
}
GET /clusters/{cluster_id}/brokers/{broker_id}/tasks

特定のブローカーのブローカータスクの一覧表示

cluster_id で指定されたクラスターの、broker_id で指定されたブローカーのすべてのブローカータスクのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。

リクエストの例

GET /clusters/{cluster_id}/brokers/{broker_id}/tasks HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    タスクのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaBrokerTaskList",
    "metadata": {
        "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/-/tasks",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaBrokerTask",
            "metadata": {
                "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1/tasks/add-broker",
                "resource_name": "crn:///kafka=cluster-1/broker=1/task=add-broker"
            },
            "cluster_id": "cluster_id",
            "broker_id": 1,
            "task_type": "add-broker",
            "task_status": "IN_PROGRESS",
            "sub_task_statuses": {
                "partition_reassignment_status": "IN_PROGRESS"
            },
            "created_at": "2019-10-12T07:20:50Z",
            "updated_at": "2019-10-12T07:20:55Z",
            "broker": {
                "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1"
            }
        },
        {
            "kind": "KafkaBrokerTask",
            "metadata": {
                "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1/tasks/remove-broker",
                "resource_name": "crn:///kafka=cluster-1/broker=1/task=remove-broker"
            },
            "cluster_id": "cluster_id",
            "broker_id": 1,
            "task_type": "remove-broker",
            "task_status": "FAILED",
            "sub_task_statuses": {
                "partition_reassignment_status": "ERROR",
                "broker_shutdown_status": "CANCELED"
            },
            "created_at": "2019-10-12T07:20:50Z",
            "updated_at": "2019-10-12T07:20:55Z",
            "error_code": 10006,
            "error_message": "Error while computing the initial remove broker plan for brokers [1] prior to shutdown.",
            "broker": {
                "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1"
            }
        }
    ]
}
GET /clusters/{cluster_id}/brokers/-/tasks/{task_type}

特定のタスクタイプのブローカータスクの一覧表示

cluster_id で指定されたクラスターの、指定された task_type を持つすべてのブローカータスクのリストを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • task_type (string) -- Kafka ブローカータスクタイプ。

リクエストの例

GET /clusters/{cluster_id}/brokers/-/tasks/{task_type} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    タスクのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaBrokerTaskList",
    "metadata": {
        "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/-/tasks",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaBrokerTask",
            "metadata": {
                "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1/tasks/add-broker",
                "resource_name": "crn:///kafka=cluster-1/broker=1/task=add-broker"
            },
            "cluster_id": "cluster_id",
            "broker_id": 1,
            "task_type": "add-broker",
            "task_status": "IN_PROGRESS",
            "sub_task_statuses": {
                "partition_reassignment_status": "IN_PROGRESS"
            },
            "created_at": "2019-10-12T07:20:50Z",
            "updated_at": "2019-10-12T07:20:55Z",
            "broker": {
                "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1"
            }
        },
        {
            "kind": "KafkaBrokerTask",
            "metadata": {
                "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/2/tasks/add-broker",
                "resource_name": "crn:///kafka=cluster-1/broker=2/task=add-broker"
            },
            "cluster_id": "cluster_id",
            "broker_id": 2,
            "task_type": "add-broker",
            "task_status": "FAILED",
            "sub_task_statuses": {
                "partition_reassignment_status": "ERROR",
                "broker_shutdown_status": "CANCELED"
            },
            "created_at": "2019-10-12T07:20:50Z",
            "updated_at": "2019-10-12T07:20:55Z",
            "error_code": 10006,
            "error_message": "Error while computing the initial add broker plan for brokers [2]",
            "broker": {
                "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/2"
            }
        }
    ]
}
GET /clusters/{cluster_id}/brokers/{broker_id}/tasks/{task_type}

1 つのブローカータスクの取得

cluster_id で指定されたクラスターの、broker_id で指定されたブローカーの task_type で指定された 1 つのブローカータスクを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。
  • task_type (string) -- Kafka ブローカータスクタイプ。

リクエストの例

GET /clusters/{cluster_id}/brokers/{broker_id}/tasks/{task_type} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    ブローカータスク

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaBrokerTask",
    "metadata": {
        "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1/tasks/add-broker",
        "resource_name": "crn:///kafka=cluster-1/broker=1/task=1"
    },
    "cluster_id": "cluster-1",
    "broker_id": 1,
    "task_type": "add-broker",
    "task_status": "FAILED",
    "sub_task_statuses": {
        "partition_reassignment_status": "ERROR"
    },
    "created_at": "2019-10-12T07:20:50Z",
    "updated_at": "2019-10-12T07:20:55Z",
    "error_code": 10013,
    "error_message": "The Confluent Balancer operation was overridden by a higher priority operation",
    "broker": {
        "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1"
    }
}

ブローカーの削除タスク

GET /clusters/{cluster_id}/remove-broker-tasks

ブローカー削除タスクの一覧表示

指定された Kafka クラスターの remove-broker-tasks のリストを返します。/remove-broker-tasks は非推奨であり、今後のリリースで削除される可能性があります。代わりに、新しい /tasks API を使用してください。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/remove-broker-tasks HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    ブローカー削除タスクのリスト。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaRemoveBrokerTaskList",
    "metadata": {
        "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/remove-broker-tasks",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaRemoveBrokerTask",
            "metadata": {
                "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/remove-broker-tasks/1",
                "resource_name": "crn:///kafka=cluster-1/remove-broker-task=1"
            },
            "cluster_id": "cluster-1",
            "broker_id": 1,
            "partition_reassignment_status": "FAILED",
            "broker_shutdown_status": "CANCELED",
            "error_code": 10006,
            "error_message": "Error while computing the initial remove broker plan for brokers [1] prior to shutdown.",
            "broker": {
                "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1"
            }
        },
        {
            "kind": "KafkaRemoveBrokerTask",
            "metadata": {
                "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/remove-broker-tasks/2",
                "resource_name": "crn:///kafka=cluster-1/remove-broker-task=2"
            },
            "cluster_id": "cluster-1",
            "broker_id": 2,
            "partition_reassignment_status": "FAILED",
            "broker_shutdown_status": "CANCELED",
            "error_code": 10006,
            "error_message": "Error while computing the initial remove broker plan for brokers [2] prior to shutdown.",
            "broker": {
                "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/2"
            }
        }
    ]
}
GET /clusters/{cluster_id}/remove-broker-tasks/{broker_id}

ブローカーの削除タスクの取得

指定された broker_id のブローカー削除タスクを返します。

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • broker_id (integer) -- Kafka ブローカー ID。

リクエストの例

GET /clusters/{cluster_id}/remove-broker-tasks/{broker_id} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    ブローカー削除タスク。

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaRemoveBrokerTask",
    "metadata": {
        "self": "http://localhost:9391/kafka/v3/clusters/cluster-1/remove-broker-tasks/1",
        "resource_name": "crn:///kafka=cluster-1/remove-broker-task=1"
    },
    "cluster_id": "cluster-1",
    "broker_id": 1,
    "partition_reassignment_status": "FAILED",
    "broker_shutdown_status": "CANCELED",
    "error_code": 10006,
    "error_message": "Error while computing the initial remove broker plan for brokers [1] prior to shutdown.",
    "broker": {
        "related": "http://localhost:9391/kafka/v3/clusters/cluster-1/brokers/1"
    }
}

Cluster Linking

GET /clusters/{cluster_id}/links

送信先クラスター内のすべてのクラスターリンクの一覧表示

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

GET /clusters/{cluster_id}/links HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    リンク名のリストとプロパティ

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaLinkDataList",
    "metadata": {
        "self": "http://localhost:63699/kafka/v3/clusters/Fds7TcT9TTqEXsoRLEKMcQ/links",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaLinkData",
            "metadata": {
                "self": "http://localhost:58441/kafka/v3/clusters/CIL-69l7S1CwoVNAhoQLug/links/my-new-link-1",
                "resource_name": null
            },
            "cluster_id": "src-cluster-id",
            "link_name": "my-new-link-1",
            "link_id": "7840644d-f7d8-4844-a577-a10ef3df31df",
            "topic_names": [
                "topic-sb-1",
                "topic-sb-2"
            ]
        },
        {
            "kind": "KafkaLinkData",
            "metadata": {
                "self": "http://localhost:58441/kafka/v3/clusters/CIL-69l7S1CwoVNAhoQLug/links/my-new-link-2",
                "resource_name": null
            },
            "cluster_id": "src-cluster-id",
            "link_name": "my-new-link-2",
            "link_id": "abced-f7d8-4844-a577-abcdefghijklmnp",
            "topic_names": [
                "topic-db-1",
                "topic-db-2"
            ]
        }
    ]
}
POST /clusters/{cluster_id}/links

クラスターリンクの作成

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。

リクエストの例

POST /clusters/{cluster_id}/links HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "source_cluster_id": "cluster-1",
    "configs": [
        {
            "name": "acl.sync.enable",
            "value": "false"
        },
        {
            "name": "consumer.offset.sync.ms",
            "value": "30000"
        }
    ]
}
ステータスコード:
GET /clusters/{cluster_id}/links/{link_name}

クラスターリンクの詳細の表示

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • link_name (string) -- リンク名

リクエストの例

GET /clusters/{cluster_id}/links/{link_name} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    1 つのリンク名とプロパティ

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaLinkData",
    "metadata": {
        "self": "http://localhost:63699/kafka/v3/clusters/Fds7TcT9TTqEXsoRLEKMcQ/links/my-new-link-1",
        "resource_name": null
    },
    "cluster_id": "src-cluster-id",
    "link_name": "my-new-link-1",
    "link_id": "7840644d-f7d8-4844-a577-a10ef3df31df",
    "topic_names": [
        "topic-db-1",
        "topic-db-2"
    ]
}
DELETE /clusters/{cluster_id}/links/{link_name}

クラスターリンクの削除

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • link_name (string) -- リンク名
ステータスコード:
  • 200 OK -- 操作は成功、応答にコンテンツなし
GET /clusters/{cluster_id}/links/{link_name}/configs

クラスターリンクのすべての構成の一覧表示

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • link_name (string) -- リンク名

リクエストの例

GET /clusters/{cluster_id}/links/{link_name}/configs HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    構成名と値

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaLinkConfigDataList",
    "metadata": {
        "self": "http://localhost:56450/kafka/v3/clusters/v-0Ce-CkTyKQol9v46LaCQ/links/link-nb-1/configs",
        "next": null
    },
    "data": [
        {
            "kind": "KafkaLinkConfigData",
            "metadata": {
                "self": "http://localhost:56608/kafka/v3/clusters/1Rh_4htxSuen7RYGvGmgNw/links/my-new-link-1",
                "resource_name": null
            },
            "cluster_id": "1Rh_4htxSuen7RYGvGmgNw",
            "name": "consumer.offset.sync.ms",
            "value": "3825940",
            "default": false,
            "read_only": false,
            "sensitive": false,
            "source": "DYNAMIC_CLUSTER_LINK_CONFIG",
            "synonyms": [
                "cosm"
            ],
            "link_name": "link-db-1",
            "link_id": "7840644d-f7d8-4844-a577-a10ef3df31df"
        },
        {
            "kind": "KafkaLinkConfigData",
            "metadata": {
                "self": "http://localhost:56608/kafka/v3/clusters/1Rh_4htxSuen7RYGvGmgNw/links/my-new-link-1",
                "resource_name": null
            },
            "cluster_id": "1Rh_4htxSuen7RYGvGmgNw",
            "name": "acl.sync.ms",
            "value": "5000",
            "default": false,
            "read_only": false,
            "sensitive": false,
            "source": "DYNAMIC_CLUSTER_LINK_CONFIG",
            "synonyms": [
                "asm"
            ],
            "link_name": "link-db-1",
            "link_id": "7840644d-f7d8-4844-a577-a10ef3df31df"
        }
    ]
}
GET /clusters/{cluster_id}/links/{link_name}/configs/{config_name}

クラスターリンク下にある構成の詳細の表示

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • link_name (string) -- リンク名
  • config_name (string) -- リンクの構成名

リクエストの例

GET /clusters/{cluster_id}/links/{link_name}/configs/{config_name} HTTP/1.1
Host: example.com
ステータスコード:
  • 200 OK --

    構成名と値

    応答の例:

    HTTP/1.1 200 OK
Content-Type: application/json

{
    "kind": "KafkaLinkConfigData",
    "metadata": {
        "self": "http://localhost:56608/kafka/v3/clusters/1Rh_4htxSuen7RYGvGmgNw/links/my-new-link-1",
        "resource_name": null
    },
    "cluster_id": "1Rh_4htxSuen7RYGvGmgNw",
    "name": "consumer.offset.sync.ms",
    "value": "3825940",
    "default": false,
    "read_only": false,
    "sensitive": false,
    "source": "DYNAMIC_CLUSTER_LINK_CONFIG",
    "synonyms": [
        "cosm"
    ],
    "link_name": "link-db-1",
    "link_id": "7840644d-f7d8-4844-a577-a10ef3df31df",
    "topics": [
        "topic-db-1",
        "topic-db-2"
    ]
}
PUT /clusters/{cluster_id}/links/{link_name}/configs/{config_name}

クラスターリンク下にある構成の変更

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • link_name (string) -- リンク名
  • config_name (string) -- リンクの構成名

リクエストの例

PUT /clusters/{cluster_id}/links/{link_name}/configs/{config_name} HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "value": "300000"
}
ステータスコード:
DELETE /clusters/{cluster_id}/links/{link_name}/configs/{config_name}

指定した構成のデフォルト値へのリセット

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • link_name (string) -- リンク名
  • config_name (string) -- リンクの構成名
ステータスコード:
POST /clusters/{cluster_id}/links/{link_name}/configs:alter

トピック構成のバッチ変更

リンク構成のバッチ変更

パラメーター:
  • cluster_id (string) -- Kafka クラスター ID。
  • link_name (string) -- リンク名

リクエストの例

POST /clusters/{cluster_id}/links/{link_name}/configs:alter HTTP/1.1
Host: example.com
Content-Type: application/json

{
    "data": [
        {
            "name": "cleanup.policy",
            "operation": "DELETE"
        },
        {
            "name": "compression.type",
            "value": "gzip"
        }
    ]
}
ステータスコード:

REST API の使用例（curl）

このセクションでは、Confluent REST API の呼び出しに curl コマンドを使用して、API エンドポイントをコマンドラインから簡単にテストする方法について説明します。以下の例では、最新の API バージョンである REST Proxy API v3 と、JSON シリアル化フォーマット（「コンテンツタイプ」を参照）を使用しています。

（REST Proxy API v2 の API 呼び出しをテストする場合は、v3 を v2 と読み替えて、"/kafka" を削除してください。REST API v2 コマンドに "kafka" を含めることはできません。v2 API のドキュメント を必ず参照してください。v3 の API の中には v2 で利用できないものがあります。）

ちなみに

アプリの場合に同じ API をソースコードから呼び出す方法については、『Confluent Admin REST APIs demo』を参照してください。

実施に際しては以下に留意してください。

  • API のテストを行う場合、curl の --silent フラグに jq を組み合わせてみてください。指定したコマンドから、きれいに書式設定された出力結果が得られます。以下の例では、このような追加の書式オプションが使用されています。
  • jq のフィルター機能は強力 ですが、curl と jq の出力を簡単な grep コマンドにパイプして、結果に対してさらにフィルターをかけられます。以下の例ではこのことも示されています。
  • API を使用して値を取得および設定するには、クラスターの URL とクラスター ID の情報が必要です。この情報は Control Center の Cluster Settings タブで確認できます（ローカルクラスターの場合はウェブブラウザーの http://localhost:9021/）。
  • 以下の例では、ローカルホスト（localhost:8090）で Kafka にアクセスするためのデフォルトのホストとポートが示されています。

既知のクラスターのリストおよび情報の表示

クラスターのリストや情報を表示するには、API エンドポイント GET /clusters を以下のように使用します。

curl --silent -X GET http://localhost:8090/kafka/v3/clusters/ | jq

例と結果:

curl --silent -X GET http://localhost:8090/kafka/v3/clusters/ | jq

"kind": "KafkaClusterList",
"metadata": {
  "self": "http://localhost:8090/kafka/v3/clusters",
  "next": null
},
"data": [
  {
    "kind": "KafkaCluster",
    "metadata": {
      "self": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg",
      "resource_name": "crn:///kafka=7cteo6omRwKaUFXj3BHxdg"
    },
    "cluster_id": "7cteo6omRwKaUFXj3BHxdg",
    "controller": {
      "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/brokers/0"
    },
    "acls": {
      "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/acls"
    },
    "brokers": {
      "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/brokers"
    },
    "broker_configs": {
      "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/broker-configs"
    },
    "consumer_groups": {
      "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/consumer-groups"
    },
    "topics": {
      "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics"
    },
    "partition_reassignments": {
      "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/-/partitions/-/reassignment"

ちなみに

現在、Kafka と REST Proxy のどちらについても、認識するのは bootstrap.servers の構成が指す Kafka クラスターに限られます。したがって、応答で Kafka クラスターは 1 つだけ返されます。

トピックの作成

トピックを作成するには、トピックエンドポイント POST /clusters/{cluster_id}/topics を以下のように使用します。

curl --silent -X POST -H "Content-Type: application/json" \
--data '{"topic_name": "<topic-name>"}' http://localhost:8090/kafka/v3/clusters/<cluster-id>/topics | jq

例と結果:

curl --silent -X POST -H "Content-Type: application/json" \
--data '{"topic_name": "my-cool-topic"}' http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics | jq

 "kind": "KafkaTopic",
 "metadata": {
   "self": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/my-cool-topic",
   "resource_name": "crn:///kafka=7cteo6omRwKaUFXj3BHxdg/topic=my-cool-topic"
 },
 "cluster_id": "7cteo6omRwKaUFXj3BHxdg",
 "topic_name": "my-cool-topic",
 "is_internal": false,
 "replication_factor": 0,
 "partitions": {
   "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/my-cool-topic/partitions"
 },
 "configs": {
   "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/my-cool-topic/configs"
 },
 "partition_reassignments": {
   "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/my-cool-topic/partitions/-/reassignment"

指定したトピックに関する情報の表示

指定したトピックの詳細情報を表示するには、トピックエンドポイント GET /clusters/{cluster_id}/topics/{topic_name} を以下のように使用します。

curl --silent -X GET http://localhost:8090/kafka/v3/clusters/<cluster-id>/topics/<topic-name> | jq

例と結果:

curl --silent -X GET http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/my-cool-topic | jq

"kind": "KafkaTopic",
 "metadata": {
   "self": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/my-cool-topic",
   "resource_name": "crn:///kafka=7cteo6omRwKaUFXj3BHxdg/topic=my-cool-topic"
 },
 "cluster_id": "7cteo6omRwKaUFXj3BHxdg",
 "topic_name": "my-cool-topic",
 "is_internal": false,
 "replication_factor": 1,
 "partitions": {
   "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/my-cool-topic/partitions"
 },
 "configs": {
   "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/my-cool-topic/configs"
 },
 "partition_reassignments": {
   "related": "http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/my-cool-topic/partitions/-/reassignment"

全トピックの一覧表示

すべてのトピック（内部トピックとユーザー作成トピック）の詳細情報を一覧表示するには、トピックエンドポイント GET /clusters/{cluster_id}/topics/ を使用します。

例:

curl --silent -X GET http://localhost:8090/kafka/v3/clusters/<cluster-id>/topics | jq

これにより、レプリケーション係数、パーティション、構成など、クラスター上の各トピックの詳細情報が表示されます。これにより、Kafka コマンド kafka-topics --describe （kafka-topics --describe --bootstrap-server localhost:9092）と同様の出力が表示されます。

全トピックの名前の一覧表示

トピックリストにフィルターを適用してトピック名のみ表示するには、エンドポイント GET /clusters/{cluster_id}/topics/ を以下のように使用します。

curl --silent -X GET http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics | jq | grep '.topic_name'

これにより、Kafka コマンド kafka-topics --list （kafka-topics --list --bootstrap-server localhost:9092）の出力と同様の情報が表示されます。

指定したプレフィックスを持つトピックの一覧表示

指定したプレフィックスを持つトピックをすべて一覧表示するには、エンドポイント GET /clusters/{cluster_id}/topics/ を以下のように使用します。

curl --silent -X GET http://localhost:8090/kafka/v3/clusters/<cluster-id>/topics | jq | grep '.topic_name' | grep '<prefix>'

例と結果:

curl --silent -X GET http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics | jq | grep '.topic_name' | grep 'my-'

"topic_name": "my-cool-topic",
"topic_name": "my-hot-topic",

トピックの削除

指定したトピックを削除するには、API エンドポイント DELETE /clusters/{cluster_id}/topics/{topic_name} を使用します。

curl --silent -X DELETE http://localhost:8090/kafka/v3/clusters/<cluster-id>/topics/{<topic-name>}

例:

curl --silent -X DELETE http://localhost:8090/kafka/v3/clusters/7cteo6omRwKaUFXj3BHxdg/topics/{my-legacy-topic} | jq

トピックが削除されたかどうかは、トピックを再び一覧表示して（または Control Center で）確認します。

ブローカータスクの一覧表示

ブローカータスクを一覧表示するには、エンドポイント GET/clusters/{cluster_id}/brokers/-/tasks に対して以下のようにクエリを実行します。この呼び出しを実行すると、Self-Balancing Clusters が有効の状態でマルチブローカークラスターを実行している場合や、大量のデータを処理中でアクティブなブローカータスクが発生している場合などに、より興味深い情報が得られます。

全ブローカーについてタスクを一覧表示する場合:

curl --silent -X GET http://localhost:8090/kafka/v3/clusters/<cluster-id>/brokers/-/tasks | jq

特定のブローカー（ここではブローカー 3）についてタスクを一覧表示する場合:

curl -X DELETE http://localhost:8090/kafka/v3/clusters/<cluster-id>/brokers/3 | jq

「Self-Balancing チュートリアル」または「Self-Balancing Clusters デモ（Docker） to experiment with Self-Balancing」を参照してください。

ブロックリスト

公開する API を制限するために、Confluent Kafka REST にブロックリストが追加されました。オペレーターは、api.endpoints.blocklist を使用して、REST Proxy インターフェイスで公開する API を簡単に選択できるようになりました。以下に例を示します。

api.endpoints.blocklist= api.v3.partitions.list,api.v3.cluster-configs.*

api.endpoints.blocklist を使用してブロックできるすべての API のリストを以下に示します。

ブロックできる API（v2 および v3）
api.v2.brokers.*
api.v2.brokers.list
api.v2.consumers.*
api.v2.consumers.assign
api.v2.consumers.commit-offsets
api.v2.consumers.consume-avro
api.v2.consumers.consume-binary
api.v2.consumers.consume-json
api.v2.consumers.consume-json-schema
api.v2.consumers.consume-protobuf
api.v2.consumers.create
api.v2.consumers.delete
api.v2.consumers.get-assignments
api.v2.consumers.get-committed-offsets
api.v2.consumers.get-subscription
api.v2.consumers.seek-to-beginning
api.v2.consumers.seek-to-end
api.v2.consumers.seek-to-offset
api.v2.consumers.subscribe
api.v2.consumers.unsubscribe
api.v2.partitions.*
api.v2.partitions.get
api.v2.partitions.get-offsets
api.v2.partitions.list
api.v2.produce-to-partition.*
api.v2.produce-to-partition.avro
api.v2.produce-to-partition.binary
api.v2.produce-to-partition.json
api.v2.produce-to-partition.json-schema
api.v2.produce-to-partition.protobuf
api.v2.produce-to-topic.*
api.v2.produce-to-topic.avro
api.v2.produce-to-topic.binary
api.v2.produce-to-topic.json
api.v2.produce-to-topic.json-schema
api.v2.produce-to-topic.protobuf
api.v2.root.*
api.v2.root.get
api.v2.root.post
api.v2.topics.*
api.v2.topics.get
api.v2.topics.list
api.v3.acls.*
api.v3.acls.create
api.v3.acls.delete
api.v3.acls.list
api.v3.broker-configs.*
api.v3.broker-configs.alter
api.v3.broker-configs.delete
api.v3.broker-configs.get
api.v3.broker-configs.list
api.v3.broker-configs.update
api.v3.brokers.*
api.v3.brokers.get
api.v3.brokers.list
api.v3.cluster-configs.*
api.v3.cluster-configs.alter
api.v3.cluster-configs.delete
api.v3.cluster-configs.get
api.v3.cluster-configs.list
api.v3.cluster-configs.update
api.v3.clusters.*
api.v3.clusters.get
api.v3.clusters.list
api.v3.consumer-assignments.*
api.v3.consumer-assignments.get
api.v3.consumer-assignments.list
api.v3.consumer-groups.*
api.v3.consumer-groups.get
api.v3.consumer-groups.list
api.v3.consumers.*
api.v3.consumers.get
api.v3.consumers.list
api.v3.partition-reassignments.*
api.v3.partition-reassignments.get
api.v3.partition-reassignments.list
api.v3.partition-reassignments.search-by-topic
api.v3.partitions.*
api.v3.partitions.get
api.v3.partitions.list
api.v3.replicas.*
api.v3.replicas.get
api.v3.replicas.list
api.v3.replicas.search-by-broker
api.v3.topic-configs.*
api.v3.topic-configs.alter
api.v3.topic-configs.delete
api.v3.topic-configs.get
api.v3.topic-configs.list
api.v3.topic-configs.update
api.v3.topics.*
api.v3.topics.create
api.v3.topics.delete
api.v3.topics.get
api.v3.topics.list