// Module included in the following assemblies:
//
// assembly-kafka-bridge-overview.adoc

[id='con-requests-kafka-bridge-{context}']
= Requests to the Kafka Bridge

[role="_abstract"]
Specify data formats and HTTP headers to ensure valid requests are submitted to the Kafka Bridge.

== Content Type headers

API request and response bodies are always encoded as JSON.

* When performing consumer operations, `POST` requests must provide the following `Content-Type` header if there is a non-empty body:
+
[source,http,subs=+quotes]
----
Content-Type: application/vnd.kafka.v2+json
----

* When performing producer operations, `POST` requests must provide `Content-Type` headers specifying the _embedded data format_ of the messages produced. This can be either `json`, `binary` or `text`.
+
[cols="35,65",options="header",stripes="none",separator=¦]
|===

¦Embedded data format
¦Content-Type header

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

¦Binary
m¦Content-Type: application/vnd.kafka.binary.v2+json

¦Text
m¦Content-Type: application/vnd.kafka.text.v2+json

|===

The embedded data format is set per consumer, as described in the next section.

The `Content-Type` must _not_ be set if the `POST` request has an empty body.
An empty body can be used to create a consumer with the default values.

== Embedded data format

The embedded data format is the format of the Kafka messages that are transmitted, over HTTP, from a producer to a consumer using the Kafka Bridge. Three embedded data formats are supported: JSON, binary and text.

When creating a consumer using the `/consumers/_groupid_` endpoint, the `POST` request body must specify an embedded data format of either JSON, binary or text. This is specified in the `format` field, for example:

[source,json,subs="attributes+"]
----
{
  "name": "my-consumer",
  "format": "binary", # <1>
  # ...
}
----
<1> A binary embedded data format.

The embedded data format specified when creating a consumer must match the data format of the Kafka messages it will consume.

If you choose to specify a binary embedded data format, subsequent producer requests must provide the binary data in the request body as Base64-encoded strings. For example, when sending messages using the `/topics/_topicname_` endpoint, `records.value` must be encoded in Base64:

[source,json,subs=attributes+]
----
{
  "records": [
    {
      "key": "my-key",
      "value": "ZWR3YXJkdGhldGhyZWVsZWdnZWRjYXQ="
    },
  ]
}
----

Producer requests must also provide a `Content-Type` header that corresponds to the embedded data format, for example, `Content-Type: application/vnd.kafka.binary.v2+json`.

== Message format

When sending messages using the `/topics` endpoint, you enter the message payload in the request body, in the `records` parameter.

The `records` parameter can contain any of these optional fields:

* Message `headers`
* Message `key`
* Message `value`
* Destination `partition`

.Example `POST` request to /topics
[source,curl,subs=attributes+]
----
curl -X POST \
  http://localhost:8080/topics/my-topic \
  -H 'content-type: application/vnd.kafka.json.v2+json' \
  -d '{
    "records": [
        {
            "key": "my-key",
            "value": "sales-lead-0001",
            "partition": 2,
            "headers": [
              {
                "key": "key1",
                "value": "QXBhY2hlIEthZmthIGlzIHRoZSBib21iIQ==" <1>
              }
            ]
        }
    ]
}'
----
<1> The header value in binary format and encoded as Base64.

Please note that if your consumer is configured to use the text embedded data format, the `value` and `key` field in the `records` parameter must be a string and not a JSON object.

== Accept headers

After creating a consumer, all subsequent GET requests must provide an `Accept` header in the following format:

[source,http,subs=+quotes]
----
Accept: application/vnd.kafka._EMBEDDED-DATA-FORMAT_.v2+json
----

The `EMBEDDED-DATA-FORMAT` is either `json`, `binary` or `text`.

For example, when retrieving records for a subscribed consumer using an embedded data format of JSON, include this Accept header:

[source,http,subs=+quotes]
----
Accept: application/vnd.kafka.json.v2+json
----

[id='con-requests-kafka-bridge-cors-{context}']
= CORS

In general, it is not possible for an HTTP client to issue requests across different domains.

For example, suppose the Kafka Bridge you deployed alongside a Kafka cluster is accessible using the `\http://my-bridge.io` domain.
HTTP clients can use the URL to interact with the Kafka Bridge and exchange messages through the Kafka cluster.
However, your client is running as a web application in the `\http://my-web-application.io` domain.
The client (source) domain is different from the Kafka Bridge (target) domain.
Because of same-origin policy restrictions, requests from the client fail.  
You can avoid this situation by using Cross-Origin Resource Sharing (CORS).

CORS allows for _simple_ and _preflighted_ requests between origin sources on different domains.

Simple requests are suitable for standard requests using `GET`, `HEAD`, `POST` methods.

A preflighted request sends a _HTTP OPTIONS_ request as an initial check that the actual request is safe to send.
On confirmation, the actual request is sent.
Preflight requests are suitable for methods that require greater safeguards, such as `PUT` and `DELETE`,
and use non-standard headers.

All requests require an _origins_ value in their header, which is the source of the HTTP request.

CORS allows you to specify allowed methods and originating URLs for accessing the Kafka cluster in your Kafka Bridge HTTP configuration.

.Example CORS configuration for Kafka Bridge
[source,properties,subs="attributes+"]
----
# ...
http.cors.enabled=true
http.cors.allowedOrigins=http://my-web-application.io
http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH
----

== Simple request

For example, this simple request header specifies the origin as `\http://my-web-application.io`.

[source,http,subs=+quotes]
----
Origin: http://my-web-application.io
----

The header information is added to the request to consume records.

[source,http,subs=+quotes]
----
curl -v -X GET _HTTP-BRIDGE-ADDRESS_/consumers/my-group/instances/my-consumer/records \
-H 'Origin: http://my-web-application.io'\
-H 'content-type: application/vnd.kafka.v2+json'
----

In the response from the Kafka Bridge, an `Access-Control-Allow-Origin` header is returned.
It contains the list of domains from where HTTP requests can be issued to the bridge.

[source,http,subs=+quotes]
----
HTTP/1.1 200 OK
Access-Control-Allow-Origin: * <1>
----
<1> Returning an asterisk (`*`) shows the resource can be accessed by any domain.

== Preflighted request

An initial preflight request is sent to Kafka Bridge using an `OPTIONS` method.
The _HTTP OPTIONS_ request sends header information to check that Kafka Bridge will allow the actual request.

Here the preflight request checks that a `POST` request is valid from `\http://my-web-application.io`.

[source,http,subs=+quotes]
----
OPTIONS /my-group/instances/my-consumer/subscription HTTP/1.1
Origin: http://my-web-application.io
Access-Control-Request-Method: POST <1>
Access-Control-Request-Headers: Content-Type <2>
----
<1> Kafka Bridge is alerted that the actual request is a `POST` request.
<2> The actual request will be sent with a `Content-Type` header.

`OPTIONS` is added to the header information of the preflight request.

[source,http,subs=+quotes]
----
curl -v -X OPTIONS -H 'Origin: http://my-web-application.io' \
-H 'Access-Control-Request-Method: POST' \
-H 'content-type: application/vnd.kafka.v2+json'
----

Kafka Bridge responds to the initial request to confirm that the request will be accepted.
The response header returns allowed origins, methods and headers.

[source,http,subs=+quotes]
----
HTTP/1.1 200 OK
Access-Control-Allow-Origin: http://my-web-application.io
Access-Control-Allow-Methods: GET,POST,PUT,DELETE,OPTIONS,PATCH
Access-Control-Allow-Headers: content-type
----

If the origin or method is rejected, an error message is returned.

The actual request does not require `Access-Control-Request-Method` header, as it was confirmed in the preflight request,
but it does require the origin header.

[source,http,subs=+quotes]
----
curl -v -X POST _HTTP-BRIDGE-ADDRESS_/topics/bridge-topic \
-H 'Origin: http://my-web-application.io' \
-H 'content-type: application/vnd.kafka.v2+json'
----

The response shows the originating URL is allowed.

[source,http,subs=+quotes]
----
HTTP/1.1 200 OK
Access-Control-Allow-Origin: http://my-web-application.io
----

[role="_additional-resources"]
.Additional resources

* link:{external-cors-link}
