
[[_paths]]
== Paths

[[_info]]
=== GET /

==== Description
Retrieves information about the Kafka Bridge instance, in JSON format.


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|Information about Kafka Bridge instance.|<<_bridgeinfo,BridgeInfo>>
|===


==== Produces

* `application/json`


==== Example HTTP response

===== Response 200
[source,json]
----
{
  "bridge_version" : "0.16.0"
}
----


[[_createconsumer]]
=== POST /consumers/{groupid}

==== Description
Creates a consumer instance in the given consumer group. You can optionally specify a consumer name and supported configuration options. It returns a base URI which must be used to construct URLs for subsequent requests against this consumer instance.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group in which to create the consumer.|string
|**Body**|**body** +
__required__|Name and configuration of the consumer. The name is unique within the scope of the consumer group. If a name is not specified, a randomly generated name is assigned. All parameters are optional. The supported configuration options are shown in the following example.|<<_consumer,Consumer>>
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|Consumer created successfully.|<<_createdconsumer,CreatedConsumer>>
|**409**|A consumer instance with the specified name already exists in the Kafka Bridge.|<<_error,Error>>
|**422**|One or more consumer configuration options have invalid values.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Consumers


==== Example HTTP request

===== Request body
[source,json]
----
{
  "name" : "consumer1",
  "format" : "binary",
  "auto.offset.reset" : "earliest",
  "enable.auto.commit" : false,
  "fetch.min.bytes" : 512,
  "consumer.request.timeout.ms" : 30000,
  "isolation.level" : "read_committed"
}
----


==== Example HTTP response

===== Response 200
[source,json]
----
{
  "instance_id" : "consumer1",
  "base_uri" : "http://localhost:8080/consumers/my-group/instances/consumer1"
}
----


===== Response 409
[source,json]
----
{
  "error_code" : 409,
  "message" : "A consumer instance with the specified name already exists in the Kafka Bridge."
}
----


===== Response 422
[source,json]
----
{
  "error_code" : 422,
  "message" : "One or more consumer configuration options have invalid values."
}
----


[[_deleteconsumer]]
=== DELETE /consumers/{groupid}/instances/{name}

==== Description
Deletes a specified consumer instance. The request for this operation MUST use the base URL (including the host and port) returned in the response from the `POST` request to `/consumers/{groupid}` that was used to create this consumer.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the consumer belongs.|string
|**Path**|**name** +
__required__|Name of the consumer to delete.|string
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|Consumer removed successfully.|No Content
|**404**|The specified consumer instance was not found.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Consumers


==== Example HTTP response

===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


[[_assign]]
=== POST /consumers/{groupid}/instances/{name}/assignments

==== Description
Assigns one or more topic partitions to a consumer.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the consumer belongs.|string
|**Path**|**name** +
__required__|Name of the consumer to assign topic partitions to.|string
|**Body**|**body** +
__required__|List of topic partitions to assign to the consumer.|<<_partitions,Partitions>>
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|Partitions assigned successfully.|No Content
|**404**|The specified consumer instance was not found.|<<_error,Error>>
|**409**|Subscriptions to topics, partitions, and patterns are mutually exclusive.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Consumers


==== Example HTTP request

===== Request body
[source,json]
----
{
  "partitions" : [ {
    "topic" : "topic",
    "partition" : 0
  }, {
    "topic" : "topic",
    "partition" : 1
  } ]
}
----


==== Example HTTP response

===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


===== Response 409
[source,json]
----
{
  "error_code" : 409,
  "message" : "Subscriptions to topics, partitions, and patterns are mutually exclusive."
}
----


[[_commit]]
=== POST /consumers/{groupid}/instances/{name}/offsets

==== Description
Commits a list of consumer offsets. To commit offsets for all records fetched by the consumer, leave the request body empty.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the consumer belongs.|string
|**Path**|**name** +
__required__|Name of the consumer.|string
|**Body**|**body** +
__optional__|List of consumer offsets to commit to the consumer offsets commit log. You can specify one or more topic partitions to commit offsets for.|<<_offsetcommitseeklist,OffsetCommitSeekList>>
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|Commit made successfully.|No Content
|**404**|The specified consumer instance was not found.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Consumers


==== Example HTTP request

===== Request body
[source,json]
----
{
  "offsets" : [ {
    "topic" : "topic",
    "partition" : 0,
    "offset" : 15
  }, {
    "topic" : "topic",
    "partition" : 1,
    "offset" : 42
  } ]
}
----


==== Example HTTP response

===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


[[_seek]]
=== POST /consumers/{groupid}/instances/{name}/positions

==== Description
Configures a subscribed consumer to fetch offsets from a particular offset the next time it fetches a set of records from a given topic partition. This overrides the default fetch behavior for consumers. You can specify one or more topic partitions.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the consumer belongs.|string
|**Path**|**name** +
__required__|Name of the subscribed consumer.|string
|**Body**|**body** +
__required__|List of partition offsets from which the subscribed consumer will next fetch records.|<<_offsetcommitseeklist,OffsetCommitSeekList>>
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|Seek performed successfully.|No Content
|**404**|The specified consumer instance was not found, or the specified consumer instance did not have one of the specified partitions assigned.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Consumers
* Seek


==== Example HTTP request

===== Request body
[source,json]
----
{
  "offsets" : [ {
    "topic" : "topic",
    "partition" : 0,
    "offset" : 15
  }, {
    "topic" : "topic",
    "partition" : 1,
    "offset" : 42
  } ]
}
----


==== Example HTTP response

===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


[[_seektobeginning]]
=== POST /consumers/{groupid}/instances/{name}/positions/beginning

==== Description
Configures a subscribed consumer to seek (and subsequently read from) the first offset in one or more given topic partitions.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the subscribed consumer belongs.|string
|**Path**|**name** +
__required__|Name of the subscribed consumer.|string
|**Body**|**body** +
__required__|List of topic partitions to which the consumer is subscribed. The consumer will seek the first offset in the specified partitions.|<<_partitions,Partitions>>
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|Seek to the beginning performed successfully.|No Content
|**404**|The specified consumer instance was not found, or the specified consumer instance did not have one of the specified partitions assigned.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Consumers
* Seek


==== Example HTTP request

===== Request body
[source,json]
----
{
  "partitions" : [ {
    "topic" : "topic",
    "partition" : 0
  }, {
    "topic" : "topic",
    "partition" : 1
  } ]
}
----


==== Example HTTP response

===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


[[_seektoend]]
=== POST /consumers/{groupid}/instances/{name}/positions/end

==== Description
Configures a subscribed consumer to seek (and subsequently read from) the offset at the end of one or more of the given topic partitions.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the subscribed consumer belongs.|string
|**Path**|**name** +
__required__|Name of the subscribed consumer.|string
|**Body**|**body** +
__optional__|List of topic partitions to which the consumer is subscribed. The consumer will seek the last offset in the specified partitions.|<<_partitions,Partitions>>
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|Seek to the end performed successfully.|No Content
|**404**|The specified consumer instance was not found, or the specified consumer instance did not have one of the specified partitions assigned.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Consumers
* Seek


==== Example HTTP request

===== Request body
[source,json]
----
{
  "partitions" : [ {
    "topic" : "topic",
    "partition" : 0
  }, {
    "topic" : "topic",
    "partition" : 1
  } ]
}
----


==== Example HTTP response

===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


[[_poll]]
=== GET /consumers/{groupid}/instances/{name}/records

==== Description
Retrieves records for a subscribed consumer, including message values, topics, and partitions. The request for this operation MUST use the base URL (including the host and port) returned in the response from the `POST` request to `/consumers/{groupid}` that was used to create this consumer.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the subscribed consumer belongs.|string
|**Path**|**name** +
__required__|Name of the subscribed consumer to retrieve records from.|string
|**Query**|**max_bytes** +
__optional__|The maximum size, in bytes, of unencoded keys and values that can be included in the response. Otherwise, an error response with code 422 is returned.|integer
|**Query**|**timeout** +
__optional__|The maximum amount of time, in milliseconds, that the HTTP Bridge spends retrieving records before timing out the request.|integer
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|Poll request executed successfully.|<<_consumerrecordlist,ConsumerRecordList>>
|**404**|The specified consumer instance was not found.|<<_error,Error>>
|**406**|The `format` used in the consumer creation request does not match the embedded format in the Accept header of this request or the bridge got a message from the topic which is not JSON encoded.|<<_error,Error>>
|**422**|Response exceeds the maximum number of bytes the consumer can receive|<<_error,Error>>
|===


==== Produces

* `application/vnd.kafka.json.v2+json`
* `application/vnd.kafka.binary.v2+json`
* `application/vnd.kafka.text.v2+json`
* `application/vnd.kafka.v2+json`


==== Tags

* Consumers


==== Example HTTP response

===== Response 200
[source,json]
----
[ {
  "topic" : "topic",
  "key" : "key1",
  "value" : {
    "foo" : "bar"
  },
  "partition" : 0,
  "offset" : 2
}, {
  "topic" : "topic",
  "key" : "key2",
  "value" : [ "foo2", "bar2" ],
  "partition" : 1,
  "offset" : 3
} ]
----

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


===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


===== Response 406
[source,json]
----
{
  "error_code" : 406,
  "message" : "The `format` used in the consumer creation request does not match the embedded format in the Accept header of this request."
}
----


===== Response 422
[source,json]
----
{
  "error_code" : 422,
  "message" : "Response exceeds the maximum number of bytes the consumer can receive"
}
----


[[_subscribe]]
=== POST /consumers/{groupid}/instances/{name}/subscription

==== Description
Subscribes a consumer to one or more topics. You can describe the topics to which the consumer will subscribe in a list (of `Topics` type) or as a `topic_pattern` field. Each call replaces the subscriptions for the subscriber.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the subscribed consumer belongs.|string
|**Path**|**name** +
__required__|Name of the consumer to subscribe to topics.|string
|**Body**|**body** +
__required__|List of topics to which the consumer will subscribe.|<<_topics,Topics>>
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|Consumer subscribed successfully.|No Content
|**404**|The specified consumer instance was not found.|<<_error,Error>>
|**409**|Subscriptions to topics, partitions, and patterns are mutually exclusive.|<<_error,Error>>
|**422**|A list (of `Topics` type) or a `topic_pattern` must be specified.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Consumers


==== Example HTTP request

===== Request body
[source,json]
----
{
  "topics" : [ "topic1", "topic2" ]
}
----


==== Example HTTP response

===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


===== Response 409
[source,json]
----
{
  "error_code" : 409,
  "message" : "Subscriptions to topics, partitions, and patterns are mutually exclusive."
}
----


===== Response 422
[source,json]
----
{
  "error_code" : 422,
  "message" : "A list (of Topics type) or a topic_pattern must be specified."
}
----


[[_listsubscriptions]]
=== GET /consumers/{groupid}/instances/{name}/subscription

==== Description
Retrieves a list of the topics to which the consumer is subscribed.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the subscribed consumer belongs.|string
|**Path**|**name** +
__required__|Name of the subscribed consumer.|string
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|List of subscribed topics and partitions.|<<_subscribedtopiclist,SubscribedTopicList>>
|**404**|The specified consumer instance was not found.|<<_error,Error>>
|===


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Consumers


==== Example HTTP response

===== Response 200
[source,json]
----
{
  "topics" : [ "my-topic1", "my-topic2" ],
  "partitions" : [ {
    "my-topic1" : [ 1, 2, 3 ]
  }, {
    "my-topic2" : [ 1 ]
  } ]
}
----


===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


[[_unsubscribe]]
=== DELETE /consumers/{groupid}/instances/{name}/subscription

==== Description
Unsubscribes a consumer from all topics.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**groupid** +
__required__|ID of the consumer group to which the subscribed consumer belongs.|string
|**Path**|**name** +
__required__|Name of the consumer to unsubscribe from topics.|string
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|Consumer unsubscribed successfully.|No Content
|**404**|The specified consumer instance was not found.|<<_error,Error>>
|===


==== Tags

* Consumers


==== Example HTTP response

===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified consumer instance was not found."
}
----


[[_healthy]]
=== GET /healthy

==== Description
Check if the bridge is running. This does not necessarily imply that it is ready to accept requests.


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|The bridge is healthy|No Content
|**500**|The bridge is not healthy|No Content
|===


[[_metrics]]
=== GET /metrics

==== Description
Retrieves the bridge metrics in Prometheus format.


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|Metrics in Prometheus format retrieved successfully.|string
|===


==== Produces

* `text/plain`


[[_openapi]]
=== GET /openapi

==== Description
Retrieves the OpenAPI v2 specification in JSON format.


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|OpenAPI v2 specification in JSON format retrieved successfully.|string
|===


==== Produces

* `application/json`


[[_ready]]
=== GET /ready

==== Description
Check if the bridge is ready and can accept requests.


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**204**|The bridge is ready|No Content
|**500**|The bridge is not ready|No Content
|===


[[_listtopics]]
=== GET /topics

==== Description
Retrieves a list of all topics.


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|List of topics.|< string > array
|===


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Topics


==== Example HTTP response

===== Response 200
[source,json]
----
[ "topic1", "topic2" ]
----


[[_send]]
=== POST /topics/{topicname}

==== Description
Sends one or more records to a given topic, optionally specifying a partition, key, or both.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**topicname** +
__required__|Name of the topic to send records to or retrieve metadata from.|string
|**Query**|**async** +
__optional__|Whether to return immediately upon sending records, instead of waiting for metadata. No offsets will be returned if specified. Defaults to false.|boolean
|**Body**|**body** +
__required__||<<_producerrecordlist,ProducerRecordList>>
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|Records sent successfully.|<<_offsetrecordsentlist,OffsetRecordSentList>>
|**404**|The specified topic was not found.|<<_error,Error>>
|**422**|The record list is not valid.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.json.v2+json`
* `application/vnd.kafka.binary.v2+json`
* `application/vnd.kafka.text.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Producer
* Topics


==== Example HTTP request

===== Request body
[source,json]
----
{
  "records" : [ {
    "key" : "key1",
    "value" : "value1"
  }, {
    "value" : "value2",
    "partition" : 1
  }, {
    "value" : "value3"
  } ]
}
----


==== Example HTTP response

===== Response 200
[source,json]
----
{
  "offsets" : [ {
    "partition" : 2,
    "offset" : 0
  }, {
    "partition" : 1,
    "offset" : 1
  }, {
    "partition" : 2,
    "offset" : 2
  } ]
}
----


===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified topic was not found."
}
----


===== Response 422
[source,json]
----
{
  "error_code" : 422,
  "message" : "The record list contains invalid records."
}
----


[[_gettopic]]
=== GET /topics/{topicname}

==== Description
Retrieves the metadata about a given topic.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**topicname** +
__required__|Name of the topic to send records to or retrieve metadata from.|string
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|Topic metadata|<<_topicmetadata,TopicMetadata>>
|===


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Topics


==== Example HTTP response

===== Response 200
[source,json]
----
{
  "name" : "topic",
  "offset" : 2,
  "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
    } ]
  } ]
}
----


[[_listpartitions]]
=== GET /topics/{topicname}/partitions

==== Description
Retrieves a list of partitions for the topic.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**topicname** +
__required__|Name of the topic to send records to or retrieve metadata from.|string
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|List of partitions|< <<_partitionmetadata,PartitionMetadata>> > array
|**404**|The specified topic was not found.|<<_error,Error>>
|===


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Topics


==== Example HTTP response

===== Response 200
[source,json]
----
[ {
  "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
  } ]
} ]
----


===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified topic was not found."
}
----


[[_sendtopartition]]
=== POST /topics/{topicname}/partitions/{partitionid}

==== Description
Sends one or more records to a given topic partition, optionally specifying a key.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**partitionid** +
__required__|ID of the partition to send records to or retrieve metadata from.|integer
|**Path**|**topicname** +
__required__|Name of the topic to send records to or retrieve metadata from.|string
|**Query**|**async** +
__optional__|Whether to return immediately upon sending records, instead of waiting for metadata. No offsets will be returned if specified. Defaults to false.|boolean
|**Body**|**body** +
__required__|List of records to send to a given topic partition, including a value (required) and a key (optional).|<<_producerrecordtopartitionlist,ProducerRecordToPartitionList>>
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|Records sent successfully.|<<_offsetrecordsentlist,OffsetRecordSentList>>
|**404**|The specified topic partition was not found.|<<_error,Error>>
|**422**|The record is not valid.|<<_error,Error>>
|===


==== Consumes

* `application/vnd.kafka.json.v2+json`
* `application/vnd.kafka.binary.v2+json`
* `application/vnd.kafka.text.v2+json`


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Producer
* Topics


==== Example HTTP request

===== Request body
[source,json]
----
{
  "records" : [ {
    "key" : "key1",
    "value" : "value1"
  }, {
    "value" : "value2"
  } ]
}
----


==== Example HTTP response

===== Response 200
[source,json]
----
{
  "offsets" : [ {
    "partition" : 2,
    "offset" : 0
  }, {
    "partition" : 1,
    "offset" : 1
  }, {
    "partition" : 2,
    "offset" : 2
  } ]
}
----


===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified topic partition was not found."
}
----


===== Response 422
[source,json]
----
{
  "error_code" : 422,
  "message" : "The record is not valid."
}
----


[[_getpartition]]
=== GET /topics/{topicname}/partitions/{partitionid}

==== Description
Retrieves partition metadata for the topic partition.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**partitionid** +
__required__|ID of the partition to send records to or retrieve metadata from.|integer
|**Path**|**topicname** +
__required__|Name of the topic to send records to or retrieve metadata from.|string
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|Partition metadata|<<_partitionmetadata,PartitionMetadata>>
|**404**|The specified topic partition was not found.|<<_error,Error>>
|===


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Topics


==== Example HTTP response

===== Response 200
[source,json]
----
{
  "partition" : 1,
  "leader" : 1,
  "replicas" : [ {
    "broker" : 1,
    "leader" : true,
    "in_sync" : true
  }, {
    "broker" : 2,
    "leader" : false,
    "in_sync" : true
  } ]
}
----


===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified topic partition was not found."
}
----


[[_getoffsets]]
=== GET /topics/{topicname}/partitions/{partitionid}/offsets

==== Description
Retrieves a summary of the offsets for the topic partition.


==== Parameters

[options="header", cols=".^2a,.^3a,.^9a,.^4a"]
|===
|Type|Name|Description|Schema
|**Path**|**partitionid** +
__required__|ID of the partition.|integer
|**Path**|**topicname** +
__required__|Name of the topic containing the partition.|string
|===


==== Responses

[options="header", cols=".^2a,.^14a,.^4a"]
|===
|HTTP Code|Description|Schema
|**200**|A summary of the offsets for the topic partition.|<<_offsetssummary,OffsetsSummary>>
|**404**|The specified topic partition was not found.|<<_error,Error>>
|===


==== Produces

* `application/vnd.kafka.v2+json`


==== Tags

* Topics


==== Example HTTP response

===== Response 200
[source,json]
----
{
  "beginning_offset" : 10,
  "end_offset" : 50
}
----


===== Response 404
[source,json]
----
{
  "error_code" : 404,
  "message" : "The specified topic partition was not found."
}
----



