Repository: strimzi/strimzi-kafka-bridge
Pull request #157 (2019-04-23 13:19:18)

Improved apicurio configuration
Added error codes + example request bodies.

tombentley: Any progress on this?
tombentley: Apparently this is awaiting review by @laidan6000.
laidan6000: Sure thing. I'm reviewing the descriptions, etc. and will let you know when done, @stanlyDoge. 
stanlyDoge: After applying your suggestions the Apicurio complains about endpoints summary lenght (which should be 120 characters max ). Do you agree to move this summary to the description?
laidan6000: Yes, I'm fine with moving the text from the "summary" field to the "description" field. The docs looks exactly the same when I preview the output in my browser.
laidan6000: For the body of each POST request, is it possible to rename "Example Value" to "Request Body"?

Also, for the new example responses, is it possible to rename "Example Value" to "Example Response"?

tombentley: One other general comment: We're using the term records here, but the rest of Strimzi uses the term messages. We should be consistent.
ppatierno: About "records" (`Record`, `RecordList` and `Records`) I took a look and we are talking about produced records and consumed records.
So `Record` is a produced record data type with just key, value and partition ... and `Records` is a list of these records sent by the producer.
The `RecordsList` is an array of consumed records with topic, key, value, partition and offset.
I would try to following this name and semantic, is it possible?
tombentley: Naming them `ProducerRecord`, `ProducerRecordList` (`ProducerRecords` is just soo easily confused with the singular version) and `ConsumerRecordList` would imho be much clearer.
ppatierno: Exactly
ppatierno: @tombentley regarding your question about commit offsets, if the request is empty it just does a commit for all the partitions related to received records (so native Kafka client means `commitSync` without parameter) if only some of them are specified, then it commits just for those partitions (so `commitSync` with a collection of TopicPartition and related OffsetAndMetadata)
tombentley: @ppatierno I wasn't asking for my own edification: Tell it to the docs!
ppatierno: @tombentley sorry it sounded as a question to me! @laidan6000 @PaulRMellor ^^
laidan6000: I suggest combining the **Topics** and **Producers** sections into one because they describe the same family of operations and the same resource. Personally, I would name it "Topics" but I don't mind either way. I think having two sections may confuse users.
ppatierno: @laidan6000 that's a good point but "Topics" it too generic because it could be applied to consumers operatin as well, not just producers. I would stick with both for now because we are going to have even admin operations in the future that will be related to topics.
ppatierno: @stanlyDoge I approved the PR but you have to fix the conflicts. Just override whatever is in the master because your openapi.json is now the source of truth
laidan6000: Please fix the typo in the GET request: ""the the Kafka Bridge instance".
ppatierno: @tombentley I think that @PaulRMellor addressed all the comments. Related to the seek errors, I opened an issue (#195) as replied to your questions. I think we can address it later because I want to test what happens at Kafka native client level. The last thing is about describing what the `base_uri` is. Do you think you have more or after that addition we can merge this one?

Patch:
diff --git a/src/main/java/io/strimzi/kafka/bridge/BridgeContentType.java b/src/main/java/io/strimzi/kafka/bridge/BridgeContentType.java
index 0b792b6..1f2e2b6 100644
--- a/src/main/java/io/strimzi/kafka/bridge/BridgeContentType.java
+++ b/src/main/java/io/strimzi/kafka/bridge/BridgeContentType.java
@@ -14,4 +14,6 @@ public class BridgeContentType {
     public static final String KAFKA_JSON_JSON = "application/vnd.kafka.json.v2+json";
     // JSON encoding with BINARY embedded format
     public static final String KAFKA_JSON_BINARY = "application/vnd.kafka.binary.v2+json";
+    // JSON encoding
+    public static final String KAFKA_JSON = "application/vnd.kafka.v2+json";
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
index 00496a6..c130281 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -380,7 +380,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
 
     private void sendConsumerCommitOffsetResponse(HttpServerResponse response, boolean result) {
         if (result) {
-            response.setStatusCode(200);
+            response.setStatusCode(204);
         } else {
             response.setStatusCode(500);
         }
diff --git a/src/main/resources/openapi.json b/src/main/resources/openapi.json
index ab9b39a..98ee1f3 100644
--- a/src/main/resources/openapi.json
+++ b/src/main/resources/openapi.json
@@ -1,314 +1,1386 @@
 {
-  "openapi": "3.0.2",
-  "info": {
-    "title": "HTTP Kafka bridge",
-    "version": "1.0.0"
-  },
-  "paths": {
-    "/topics/{topicname}": {
-      "post": {
-        "description": "Write to topic",
-        "operationId": "send",
-        "responses": {
-          "200": {
-            "description": "OK",
-            "content": {
-              "application/json": {}
-            }
-          }
-        }
-      },
-      "parameters": [
-        {
-          "name": "topicname",
-          "in": "path",
-          "description": "Topic name",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
-        }
-      ]
-    },
-    "/topics/{topicname}/partitions/{partitionid}": {
-      "post": {
-        "operationId": "sendToPartition",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
-        }
-      },
-      "parameters": [
-        {
-          "name": "topicname",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+    "openapi": "3.0.0",
+    "info": {
+        "title": "Strimzi HTTP Bridge for Apache Kafka",
+        "description": "The Strimzi HTTP Bridge for Apache Kafka provides a REST API for integrating HTTP based client applications with a Kafka cluster. You can use the API to create and manage consumers and send and receive records over HTTP rather than the native Kafka protocol. ",
+        "termsOfService": "http://swagger.io/terms/",
+        "contact": {
+            "email": "strimzi@redhat.com"
         },
-        {
-          "name": "partitionid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "integer"
-          }
-        }
-      ]
+        "license": {
+            "name": "Apache 2.0",
+            "url": "http://www.apache.org/licenses/LICENSE-2.0.html"
+        },
+        "version": "0.1.0"
     },
-    "/consumers/{groupid}": {
-      "post": {
-        "operationId": "createConsumer",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
-        }
-      },
-      "parameters": [
+    "servers": [
         {
-          "name": "groupid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
-        }
-      ]
-    },
-    "/consumers/{groupid}/instances/{name}": {
-      "delete": {
-        "operationId": "deleteConsumer",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
+            "url": "http://bridge.swagger.io/v2"
         }
-      },
-      "parameters": [
-        {
-          "name": "groupid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+    ],
+    "paths": {
+        "/consumers/{groupid}": {
+            "post": {
+                "tags": [
+                    "Consumers"
+                ],
+                "description": "Creates a consumer instance in the given consumer group. You can optionally specify a consumer name and supported configuration options. It resturns a base URI used to construct URIs for subsequent requests against this consumer instance.",
+                "operationId": "createConsumer",
+                "requestBody": {
+                    "description": "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.",
+                    "content": {
+                        "application/vnd.kafka.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/Consumer"
+                            }
+                        }
+                    },
+                    "required": true
+                },
+                "responses": {
+                    "200": {
+                        "description": "Consumer created successfully.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/CreatedConsumer"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "instance_id": "consumer1",
+                                            "base_uri": "http://localhost:8080/consumers/my-group/instances/consumer1"
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    },
+                    "409": {
+                        "description": "A consumer instance with the specified name already exists in the Kafka Bridge.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 409,
+                                            "message": "A consumer instance with the specified name already exists in the Kafka Bridge."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    },
+                    "422": {
+                        "description": "One or more consumer configuration options have invalid values.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 422,
+                                            "message": "One or more consumer configuration options have invalid values."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "groupid",
+                    "in": "path",
+                    "description": "ID of the consumer group in which to create the consumer.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
         },
-        {
-          "name": "name",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
-        }
-      ]
-    },
-    "/consumers/{groupid}/instances/{name}/subscription": {
-      "post": {
-        "operationId": "subscribe",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
-        }
-      },
-      "delete": {
-        "operationId": "unsubscribe",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
-        }
-      },
-      "parameters": [
-        {
-          "name": "groupid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+        "/consumers/{groupid}/instances/{name}/positions/beginning": {
+            "post": {
+                "tags": [
+                    "Seek",
+                    "Consumers"
+                ],
+                "description": "Configures a subscribed consumer to seek (and subsequently read from) the first offset in one or more given topic partitions.",
+                "operationId": "seekToBeginning",
+                "requestBody": {
+                    "description": "List of topic partitions to which the consumer is subscribed. The consumer will seek the first offset in the specified partitions.",
+                    "content": {
+                        "application/vnd.kafka.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/Partitions"
+                            }
+                        }
+                    },
+                    "required": true
+                },
+                "responses": {
+                    "204": {
+                        "description": "Seek to the beginning performed successfully."
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "groupid",
+                    "in": "path",
+                    "description": "ID of the consumer group to which the subscribed consumer belongs.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "name",
+                    "in": "path",
+                    "description": "Name of the subscribed consumer.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
         },
-        {
-          "name": "name",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
-        }
-      ]
-    },
-    "/consumers/{groupid}/instances/{name}/records": {
-      "get": {
-        "operationId": "poll",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
-        }
-      },
-      "parameters": [
-        {
-          "name": "groupid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+        "/consumers/{groupid}/instances/{name}/positions/end": {
+            "post": {
+                "tags": [
+                    "Seek",
+                    "Consumers"
+                ],
+                "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.",
+                "operationId": "seekToEnd",
+                "requestBody": {
+                    "description": "List of topic partitions to which the consumer is subscribed. The consumer will seek the last offset in the specified partitions.",
+                    "content": {
+                        "application/vnd.kafka.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/Partitions"
+                            }
+                        }
+                    },
+                    "required": true
+                },
+                "responses": {
+                    "204": {
+                        "description": "Seek to the end performed successfully."
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "groupid",
+                    "in": "path",
+                    "description": "ID of the consumer group to which the subscribed consumer belongs.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "name",
+                    "in": "path",
+                    "description": "Name of the subscribed consumer.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
         },
-        {
-          "name": "name",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
-        }
-      ]
-    },
-    "/consumers/{groupid}/instances/{name}/offsets": {
-      "post": {
-        "operationId": "commit",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
-        }
-      },
-      "parameters": [
-        {
-          "name": "groupid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+        "/consumers/{groupid}/instances/{name}/subscription": {
+            "post": {
+                "tags": [
+                    "Consumers"
+                ],
+                "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.",
+                "operationId": "subscribe",
+                "requestBody": {
+                    "description": "List of topics to which the consumer will subscribe.",
+                    "content": {
+                        "application/vnd.kafka.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/Topics"
+                            }
+                        }
+                    },
+                    "required": true
+                },
+                "responses": {
+                    "204": {
+                        "description": "Consumer subscribed successfully."
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    },
+                    "409": {
+                        "description": "Subscription to topics, partitions, and patterns are mutually exclusive.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 409,
+                                            "message": "Subscription to topics, partitions, and patterns are mutually exclusive."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "delete": {
+                "tags": [
+                    "Consumers"
+                ],
+                "description": "Unsubscribes a consumer from all topics.",
+                "operationId": "unsubscribe",
+                "responses": {
+                    "204": {
+                        "description": "Consumer unsubscribed successfully."
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "groupid",
+                    "in": "path",
+                    "description": "ID of the consumer group to which the subscribed consumer belongs.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "name",
+                    "in": "path",
+                    "description": "Name of the consumer that you want to unsubscribe from topics.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
         },
-        {
-          "name": "name",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
-        }
-      ]
-    },
-    "/consumers/{groupid}/instances/{name}/positions": {
-      "post": {
-        "operationId": "seek",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
-        }
-      },
-      "parameters": [
-        {
-          "name": "groupid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+        "/consumers/{groupid}/instances/{name}/positions": {
+            "post": {
+                "tags": [
+                    "Seek",
+                    "Consumers"
+                ],
+                "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.",
+                "operationId": "seek",
+                "requestBody": {
+                    "description": "List of partition offsets from which the subscribed consumer will next fetch records.",
+                    "content": {
+                        "application/vnd.kafka.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/OffsetCommitSeekList"
+                            }
+                        }
+                    },
+                    "required": true
+                },
+                "responses": {
+                    "204": {
+                        "description": "Seek performed successfully."
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "groupid",
+                    "in": "path",
+                    "description": "ID of the consumer group to which the consumer belongs.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "name",
+                    "in": "path",
+                    "description": "Name of the subscribed consumer.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
         },
-        {
-          "name": "name",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
-        }
-      ]
-    },
-    "/consumers/{groupid}/instances/{name}/positions/beginning": {
-      "post": {
-        "operationId": "seekToBeginning",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
-        }
-      },
-      "parameters": [
-        {
-          "name": "groupid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+        "/consumers/{groupid}/instances/{name}/assignments": {
+            "post": {
+                "tags": [
+                    "Consumers"
+                ],
+                "description": "Assigns one or more topic partitions to a consumer.",
+                "operationId": "assign",
+                "requestBody": {
+                    "description": "List of topic partitions to assign to the consumer.",
+                    "content": {
+                        "application/vnd.kafka.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/Partitions"
+                            }
+                        }
+                    },
+                    "required": true
+                },
+                "responses": {
+                    "204": {
+                        "description": "Partitions assigned successfully."
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    },
+                    "409": {
+                        "description": "Subscription to topics, partitions, and patterns are mutually exclusive.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 409,
+                                            "message": "Subscription to topics, partitions, and patterns are mutually exclusive."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "groupid",
+                    "in": "path",
+                    "description": "ID of the consumer group to which the consumer belongs.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "name",
+                    "in": "path",
+                    "description": "Name of the consumer to which you want to assign topic partitions.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
         },
-        {
-          "name": "name",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+        "/topics/{topicname}": {
+            "post": {
+                "tags": [
+                    "Topics",
+                    "Producer"
+                ],
+                "description": "Sends one or more records to a given topic, optionally specifying a partition, key, or both.",
+                "operationId": "send",
+                "requestBody": {
+                    "content": {
+                        "application/vnd.kafka.json.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/ProducerRecordList"
+                            }
+                        },
+                        "application/vnd.kafka.binary.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/ProducerRecordList"
+                            }
+                        }
+                    },
+                    "required": true
+                },
+                "responses": {
+                    "200": {
+                        "description": "Records sent successfully.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/OffsetRecordSentList"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "offsets": [
+                                                {
+                                                    "partition": 2,
+                                                    "offset": 0
+                                                },
+                                                {
+                                                    "partition": 1,
+                                                    "offset": 1
+                                                },
+                                                {
+                                                    "partition": 2,
+                                                    "offset": 2
+                                                }
+                                            ]
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    },
+                    "404": {
+                        "description": "The specified topic was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified topic was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    },
+                    "422": {
+                        "description": "The record list is not valid.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 422,
+                                            "message": "The record list contains invalid records."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "topicname",
+                    "in": "path",
+                    "description": "Name of the topic to which you want to send records.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
+        },
+        "/consumers/{groupid}/instances/{name}/records": {
+            "get": {
+                "tags": [
+                    "Consumers"
+                ],
+                "description": "Retrieves records for a subscribed consumer, including message values, topics, and partitions. The operation MUST be routed to the Kafka Bridge instance that contains the consumer.",
+                "operationId": "poll",
+                "responses": {
+                    "200": {
+                        "description": "Poll request executed successfully.",
+                        "content": {
+                            "application/vnd.kafka.json.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/ConsumerRecordList"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": [
+                                            {
+                                                "topic": "topic",
+                                                "key": "key1",
+                                                "value": {
+                                                    "foo": "bar"
+                                                },
+                                                "partition": 0,
+                                                "offset": 2
+                                            },
+                                            {
+                                                "topic": "topic",
+                                                "key": "key2",
+                                                "value": [
+                                                    "foo2",
+                                                    "bar2"
+                                                ],
+                                                "partition": 1,
+                                                "offset": 3
+                                            }
+                                        ]
+                                    }
+                                }
+                            },
+                            "application/vnd.kafka.binary.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/ConsumerRecordList"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": "[\n  {\n    \"topic\": \"test\",\n    \"key\": \"a2V5\",\n    \"value\": \"Y29uZmx1ZW50\",\n    \"partition\": 1,\n    \"offset\": 100,\n  },\n  {\n    \"topic\": \"test\",\n    \"key\": \"a2V5\",\n    \"value\": \"a2Fma2E=\",\n    \"partition\": 2,\n    \"offset\": 101,\n  }\n]"
+                                    }
+                                }
+                            },
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/ConsumerRecordList"
+                                }
+                            }
+                        }
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/vnd.kafka.json.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                }
+                            },
+                            "application/vnd.kafka.binary.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                }
+                            },
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    },
+                    "406": {
+                        "description": "The `format` used in the consumer creation request does not match the embedded format in the Accept header of this request.",
+                        "content": {
+                            "application/vnd.kafka.json.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                }
+                            },
+                            "application/vnd.kafka.binary.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                }
+                            },
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "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."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "groupid",
+                    "in": "path",
+                    "description": "ID of the consumer group to which the subscribed consumer belongs.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "name",
+                    "in": "path",
+                    "description": "Name of the subscribed consumer for which you want to retrieve records.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "timeout",
+                    "in": "query",
+                    "description": "The maximum amount of time, in milliseconds, that the HTTP Bridge spends retrieving records before timing out the request.",
+                    "required": false,
+                    "schema": {
+                        "type": "integer"
+                    }
+                },
+                {
+                    "name": "max_bytes",
+                    "in": "query",
+                    "description": "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.",
+                    "required": false,
+                    "schema": {
+                        "type": "integer"
+                    }
+                }
+            ]
+        },
+        "/topics/{topicname}/partitions/{partitionid}": {
+            "post": {
+                "tags": [
+                    "Topics",
+                    "Producer"
+                ],
+                "description": "Sends one or more records to a given topic partition, optionally specifying a key.",
+                "operationId": "sendToPartition",
+                "requestBody": {
+                    "description": "List of records to send to a given topic partition, including a value (required) and a key (optional).",
+                    "content": {
+                        "application/vnd.kafka.json.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/ProducerRecordToPartitionList"
+                            }
+                        },
+                        "application/vnd.kafka.binary.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/ProducerRecordToPartitionList"
+                            }
+                        }
+                    },
+                    "required": true
+                },
+                "responses": {
+                    "200": {
+                        "description": "Records sent successfully.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/OffsetRecordSentList"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "offsets": [
+                                                {
+                                                    "partition": 2,
+                                                    "offset": 0
+                                                },
+                                                {
+                                                    "partition": 1,
+                                                    "offset": 1
+                                                },
+                                                {
+                                                    "partition": 2,
+                                                    "offset": 2
+                                                }
+                                            ]
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    },
+                    "404": {
+                        "description": "The specified topic partition was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified topic partition was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    },
+                    "422": {
+                        "description": "The record is not valid.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 422,
+                                            "message": "The record is not valid."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "topicname",
+                    "in": "path",
+                    "description": "Name of the topic containing the partition to which you want to send records.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "partitionid",
+                    "in": "path",
+                    "description": "ID of the partition to which you want to send records.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
+        },
+        "/consumers/{groupid}/instances/{name}": {
+            "delete": {
+                "tags": [
+                    "Consumers"
+                ],
+                "description": "Deletes a specified consumer instance. The operation MUST be routed to the Kafka Bridge instance that contains the consumer.",
+                "operationId": "deleteConsumer",
+                "responses": {
+                    "204": {
+                        "description": "Consumer removed successfully."
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "groupid",
+                    "in": "path",
+                    "description": "ID of the consumer group to which the consumer belongs.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "name",
+                    "in": "path",
+                    "description": "Name of the consumer that you want to delete.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
+        },
+        "/consumers/{groupid}/instances/{name}/offsets": {
+            "post": {
+                "tags": [
+                    "Consumers"
+                ],
+                "description": "Commits a list of consumer offsets. To commit offsets for all records fetched by the consumer, leave the request body empty.",
+                "operationId": "commit",
+                "requestBody": {
+                    "description": "List of consumer offsets to commit to the consumer offsets commit log. You can specify one or more topic partitions to commit offsets for.",
+                    "content": {
+                        "application/vnd.kafka.v2+json": {
+                            "schema": {
+                                "$ref": "#/components/schemas/OffsetCommitSeekList"
+                            }
+                        }
+                    }
+                },
+                "responses": {
+                    "204": {
+                        "description": "Commit made successfully."
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            },
+            "parameters": [
+                {
+                    "name": "groupid",
+                    "in": "path",
+                    "description": "ID of the consumer group to which the consumer belongs.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                },
+                {
+                    "name": "name",
+                    "in": "path",
+                    "description": "Name of the consumer.",
+                    "required": true,
+                    "schema": {
+                        "type": "string"
+                    }
+                }
+            ]
         }
-      ]
     },
-    "/consumers/{groupid}/instances/{name}/positions/end": {
-      "post": {
-        "operationId": "seekToEnd",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
+    "components": {
+        "schemas": {
+            "Consumer": {
+                "title": "Consumer",
+                "type": "object",
+                "properties": {
+                    "name": {
+                        "description": "The unique name for the consumer instance. The name is unique within the scope of the consumer group. The name is used in URLs. ",
+                        "type": "string"
+                    },
+                    "format": {
+                        "description": "The allowable message format for the consumer, which can be `binary` (default) or `json`. The messages are converted into a JSON format. ",
+                        "type": "string"
+                    },
+                    "auto.offset.reset": {
+                        "description": "Resets the offset position for the consumer. \nIf set to `earliest`, messages are read from the first offset. \nIf set to `latest`, messages are read from the latest offset.",
+                        "type": "string"
+                    },
+                    "auto.commit.enable": {
+                        "description": "If set to `true`, message offsets are committed automatically for the consumer. If set to `false`, message offsets must be committed manually.",
+                        "type": "string"
+                    },
+                    "fetch.min.bytes": {
+                        "description": "Sets the minimum ammount of data in bytes for the consumer to receive. The broker waits until the data to send exceeds this amount.",
+                        "type": "string"
+                    },
+                    "consumer.request.timeout.ms": {
+                        "description": "Sets the maximum amount of time, in milliseconds, for the consumer to wait for messages for a request. If the timeout period is reached without a response, an error is returned.",
+                        "type": "string"
+                    }
+                },
+                "example": {
+                    "name": "consumer1",
+                    "format": "binary",
+                    "auto.offset.reset": "earliest",
+                    "auto.commit.enable": "false",
+                    "fetch.min.bytes": "512",
+                    "consumer.request.timeout.ms": "30000"
+                }
+            },
+            "Partition": {
+                "title": "Partition",
+                "type": "object",
+                "properties": {
+                    "partition": {
+                        "format": "int32",
+                        "type": "integer"
+                    },
+                    "topic": {
+                        "type": "string"
+                    }
+                }
+            },
+            "Topics": {
+                "title": "Topics",
+                "type": "object",
+                "properties": {
+                    "topics": {
+                        "type": "array",
+                        "items": {
+                            "type": "string"
+                        }
+                    }
+                },
+                "example": {
+                    "topics": [
+                        "topic1",
+                        "topic2"
+                    ]
+                }
+            },
+            "Partitions": {
+                "title": "Partitions",
+                "type": "object",
+                "properties": {
+                    "partitions": {
+                        "type": "array",
+                        "items": {
+                            "$ref": "#/components/schemas/Partition"
+                        }
+                    }
+                },
+                "example": {
+                    "partitions": [
+                        {
+                            "topic": "topic",
+                            "partition": 0
+                        },
+                        {
+                            "topic": "topic",
+                            "partition": 1
+                        }
+                    ]
+                }
+            },
+            "ConsumerRecordList": {
+                "title": "ConsumerRecordList",
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "topic": {
+                            "type": "string"
+                        },
+                        "key": {
+                            "type": "string"
+                        },
+                        "value": {
+                            "type": "string"
+                        },
+                        "partition": {
+                            "format": "int32",
+                            "type": "integer"
+                        },
+                        "offset": {
+                            "format": "int64",
+                            "type": "long"
+                        }
+                    }
+                },
+                "example": [
+                    {
+                        "topic": "topic",
+                        "key": "key1",
+                        "value": "value1",
+                        "partition": 0,
+                        "offset": 2
+                    },
+                    {
+                        "topic": "topic",
+                        "key": "key2",
+                        "value": "value2",
+                        "partition": 1,
+                        "offset": 3
+                    }
+                ]
+            },
+            "CreatedConsumer": {
+                "title": "CreatedConsumer",
+                "type": "object",
+                "properties": {
+                    "instance_id": {
+                        "description": "Unique ID for the consumer instance in the group.",
+                        "type": "string"
+                    },
+                    "base_uri": {
+                        "description": "Base URI used to construct URIs for subsequent requests against this consumer instance.",
+                        "type": "string"
+                    }
+                },
+                "example": {
+                    "instance_id": "my-consumer",
+                    "base_uri": "http://localhost:8080/consumers/my-group/instances/my-consumer"
+                }
+            },
+            "OffsetCommitSeekList": {
+                "title": "OffsetCommitSeekList",
+                "type": "object",
+                "properties": {
+                    "offsets": {
+                        "type": "array",
+                        "items": {
+                            "$ref": "#/components/schemas/OffsetCommitSeek"
+                        }
+                    }
+                },
+                "example": {
+                    "offsets": [
+                        {
+                            "topic": "topic",
+                            "partition": 0,
+                            "offset": 15
+                        },
+                        {
+                            "topic": "topic",
+                            "partition": 1,
+                            "offset": 42
+                        }
+                    ]
+                }
+            },
+            "OffsetCommitSeek": {
+                "title": "OffsetCommitSeek",
+                "required": [],
+                "type": "object",
+                "properties": {
+                    "partition": {
+                        "format": "int32",
+                        "type": "integer"
+                    },
+                    "offset": {
+                        "format": "int64",
+                        "type": "integer"
+                    },
+                    "topic": {
+                        "type": "string"
+                    }
+                },
+                "example": {
+                    "partition": 43,
+                    "offset": 92,
+                    "topic": "topic"
+                }
+            },
+            "OffsetRecordSent": {
+                "title": "OffsetRecordSent",
+                "type": "object",
+                "properties": {
+                    "partition": {
+                        "format": "int32",
+                        "type": "integer"
+                    },
+                    "offset": {
+                        "format": "int64",
+                        "type": "integer"
+                    }
+                },
+                "example": {
+                    "partition": 31,
+                    "offset": 86
+                }
+            },
+            "OffsetRecordSentList": {
+                "title": "OffsetRecordSentList",
+                "type": "object",
+                "properties": {
+                    "offsets": {
+                        "type": "array",
+                        "items": {
+                            "$ref": "#/components/schemas/OffsetRecordSent"
+                        }
+                    }
+                },
+                "example": {
+                    "offsets": [
+                        {
+                            "partition": 92,
+                            "offset": 98
+                        },
+                        {
+                            "partition": 65,
+                            "offset": 91
+                        }
+                    ]
+                }
+            },
+            "Error": {
+                "title": "Error",
+                "type": "object",
+                "properties": {
+                    "error_code": {
+                        "format": "int32",
+                        "type": "integer"
+                    },
+                    "message": {
+                        "type": "string"
+                    }
+                },
+                "example": "{\n    \"error_code\": 404,\n    \"message\": \"resource not found\"\n}"
+            },
+            "ProducerRecord": {
+                "title": "ProducerRecord",
+                "required": [
+                    "value"
+                ],
+                "type": "object",
+                "properties": {
+                    "partition": {
+                        "format": "int32",
+                        "type": "integer"
+                    },
+                    "value": {
+                        "oneOf": [
+                            {
+                                "type": "object"
+                            },
+                            {
+                                "type": "string"
+                            }
+                        ]
+                    },
+                    "key": {
+                        "oneOf": [
+                            {
+                                "type": "object"
+                            },
+                            {
+                                "type": "string"
+                            }
+                        ]
+                    }
+                },
+                "example": "{\n    \"key\": \"key\"\n    \"partition\": 23\n    \"value\": {\"foo\": \"bar\"}\n}"
+            },
+            "ProducerRecordList": {
+                "title": "ProducerRecordList",
+                "type": "object",
+                "properties": {
+                    "records": {
+                        "type": "array",
+                        "items": {
+                            "$ref": "#/components/schemas/ProducerRecord"
+                        }
+                    }
+                },
+                "example": {
+                    "records": [
+                        {
+                            "key": "key1",
+                            "value": "value1"
+                        },
+                        {
+                            "value": "value2",
+                            "partition": 1
+                        },
+                        {
+                            "value": "value3"
+                        }
+                    ]
+                }
+            },
+            "ProducerRecordToPartition": {
+                "title": "ProducerRecordToPartition",
+                "required": [
+                    "value"
+                ],
+                "type": "object",
+                "properties": {
+                    "value": {
+                        "oneOf": [
+                            {
+                                "type": "object"
+                            },
+                            {
+                                "type": "string"
+                            }
+                        ]
+                    },
+                    "key": {
+                        "oneOf": [
+                            {
+                                "type": "object"
+                            },
+                            {
+                                "type": "string"
+                            }
+                        ]
+                    }
+                },
+                "example": {
+                    "value": {
+                        "v": 128
+                    },
+                    "key": {
+                        "k": "key"
+                    }
+                }
+            },
+            "ProducerRecordToPartitionList": {
+                "title": "ProducerRecordToPartitionList",
+                "type": "object",
+                "properties": {
+                    "records": {
+                        "type": "array",
+                        "items": {
+                            "$ref": "#/components/schemas/ProducerRecordToPartition"
+                        }
+                    }
+                },
+                "example": {
+                    "records": [
+                        {
+                            "key": "key1",
+                            "value": "value1"
+                        },
+                        {
+                            "value": "value2"
+                        }
+                    ]
+                }
+            }
         }
-      },
-      "parameters": [
+    },
+    "tags": [
         {
-          "name": "groupid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+            "name": "Consumers",
+            "description": "Consumer operations to create consumers in your Kafka cluster and perform common actions, such as subscribing to topics, retrieving processed records, and committing offsets."
         },
         {
-          "name": "name",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
-        }
-      ]
-    },
-    "/consumers/{groupid}/instances/{name}/assignments": {
-      "post": {
-        "operationId": "assign",
-        "responses": {
-          "200": {
-            "description": "OK"
-          }
-        }
-      },
-      "parameters": [
+            "name": "Topics",
+            "description": "Topic operations to send messages to a specified topic or topic partition. You can include message keys in requests."
+        },
         {
-          "name": "groupid",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+            "name": "Seek",
+            "description": "Seek operations that enable a consumer to begin receiving messages from a given offset position."
         },
         {
-          "name": "name",
-          "in": "path",
-          "required": true,
-          "schema": {
-            "type": "string"
-          }
+            "name": "Producer",
+            "description": "Producer operations to send records to a specified topic or topic partition."
         }
-      ]
+    ],
+    "externalDocs": {
+        "description": "Find out more about the Strimzi HTTP Bridge",
+        "url": "https://strimzi.io/documentation/"
     }
-  }
 }
\ No newline at end of file
diff --git a/src/main/resources/openapiv2.json b/src/main/resources/openapiv2.json
new file mode 100644
index 0000000..ccb6708
--- /dev/null
+++ b/src/main/resources/openapiv2.json
@@ -0,0 +1,1222 @@
+{
+  "swagger": "2.0",
+  "info": {
+    "title": "Strimzi HTTP Bridge for Apache Kafka",
+    "description": "The Strimzi HTTP Bridge for Apache Kafka provides a REST API for integrating HTTP based client applications with a Kafka cluster. You can use the API to create and manage consumers and send and receive records over HTTP rather than the native Kafka protocol. ",
+    "termsOfService": "http://swagger.io/terms/",
+    "contact": {
+      "email": "strimzi@redhat.com"
+    },
+    "license": {
+      "name": "Apache 2.0",
+      "url": "http://www.apache.org/licenses/LICENSE-2.0.html"
+    },
+    "version": "0.1.0"
+  },
+  "host": "bridge.swagger.io",
+  "basePath": "/v2",
+  "schemes": [
+    "http"
+  ],
+  "consumes": [
+    "application/json"
+  ],
+  "produces": [
+    "application/json"
+  ],
+  "paths": {
+    "/consumers/{groupid}": {
+      "post": {
+        "tags": [
+          "Consumers"
+        ],
+        "description": "Creates a consumer instance in the given consumer group. You can optionally specify a consumer name and supported configuration options. It resturns a base URI used to construct URIs for subsequent requests against this consumer instance.",
+        "operationId": "createConsumer",
+        "consumes": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "description": "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.",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/Consumer"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Consumer created successfully.",
+            "schema": {
+              "$ref": "#/definitions/CreatedConsumer"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "instance_id": "consumer1",
+                "base_uri": "http://localhost:8080/consumers/my-group/instances/consumer1"
+              }
+            }
+          },
+          "409": {
+            "description": "A consumer instance with the specified name already exists in the Kafka Bridge.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 409,
+                "message": "A consumer instance with the specified name already exists in the Kafka Bridge."
+              }
+            }
+          },
+          "422": {
+            "description": "One or more consumer configuration options have invalid values.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 422,
+                "message": "One or more consumer configuration options have invalid values."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "groupid",
+          "in": "path",
+          "description": "ID of the consumer group in which to create the consumer.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    },
+    "/consumers/{groupid}/instances/{name}/positions/beginning": {
+      "post": {
+        "tags": [
+          "Seek",
+          "Consumers"
+        ],
+        "description": "Configures a subscribed consumer to seek (and subsequently read from) the first offset in one or more given topic partitions.",
+        "operationId": "seekToBeginning",
+        "consumes": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "description": "List of topic partitions to which the consumer is subscribed. The consumer will seek the first offset in the specified partitions.",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/Partitions"
+            }
+          }
+        ],
+        "responses": {
+          "204": {
+            "description": "Seek to the beginning performed successfully."
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "groupid",
+          "in": "path",
+          "description": "ID of the consumer group to which the subscribed consumer belongs.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "name",
+          "in": "path",
+          "description": "Name of the subscribed consumer.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    },
+    "/consumers/{groupid}/instances/{name}/positions/end": {
+      "post": {
+        "tags": [
+          "Seek",
+          "Consumers"
+        ],
+        "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.",
+        "operationId": "seekToEnd",
+        "consumes": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "description": "List of topic partitions to which the consumer is subscribed. The consumer will seek the last offset in the specified partitions.",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/Partitions"
+            }
+          }
+        ],
+        "responses": {
+          "204": {
+            "description": "Seek to the end performed successfully."
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "groupid",
+          "in": "path",
+          "description": "ID of the consumer group to which the subscribed consumer belongs.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "name",
+          "in": "path",
+          "description": "Name of the subscribed consumer.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    },
+    "/consumers/{groupid}/instances/{name}/subscription": {
+      "post": {
+        "tags": [
+          "Consumers"
+        ],
+        "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.",
+        "operationId": "subscribe",
+        "consumes": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "description": "List of topics to which the consumer will subscribe.",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/Topics"
+            }
+          }
+        ],
+        "responses": {
+          "204": {
+            "description": "Consumer subscribed successfully."
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          },
+          "409": {
+            "description": "Subscription to topics, partitions, and patterns are mutually exclusive.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 409,
+                "message": "Subscription to topics, partitions, and patterns are mutually exclusive."
+              }
+            }
+          }
+        }
+      },
+      "delete": {
+        "tags": [
+          "Consumers"
+        ],
+        "description": "Unsubscribes a consumer from all topics.",
+        "operationId": "unsubscribe",
+        "responses": {
+          "204": {
+            "description": "Consumer unsubscribed successfully."
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "groupid",
+          "in": "path",
+          "description": "ID of the consumer group to which the subscribed consumer belongs.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "name",
+          "in": "path",
+          "description": "Name of the consumer that you want to unsubscribe from topics.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    },
+    "/consumers/{groupid}/instances/{name}/positions": {
+      "post": {
+        "tags": [
+          "Seek",
+          "Consumers"
+        ],
+        "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.",
+        "operationId": "seek",
+        "consumes": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "description": "List of partition offsets from which the subscribed consumer will next fetch records.",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/OffsetCommitSeekList"
+            }
+          }
+        ],
+        "responses": {
+          "204": {
+            "description": "Seek performed successfully."
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "groupid",
+          "in": "path",
+          "description": "ID of the consumer group to which the consumer belongs.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "name",
+          "in": "path",
+          "description": "Name of the subscribed consumer.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    },
+    "/consumers/{groupid}/instances/{name}/assignments": {
+      "post": {
+        "tags": [
+          "Consumers"
+        ],
+        "description": "Assigns one or more topic partitions to a consumer.",
+        "operationId": "assign",
+        "consumes": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "description": "List of topic partitions to assign to the consumer.",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/Partitions"
+            }
+          }
+        ],
+        "responses": {
+          "204": {
+            "description": "Partitions assigned successfully."
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          },
+          "409": {
+            "description": "Subscription to topics, partitions, and patterns are mutually exclusive.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 409,
+                "message": "Subscription to topics, partitions, and patterns are mutually exclusive."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "groupid",
+          "in": "path",
+          "description": "ID of the consumer group to which the consumer belongs.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "name",
+          "in": "path",
+          "description": "Name of the consumer to which you want to assign topic partitions.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    },
+    "/topics/{topicname}": {
+      "post": {
+        "tags": [
+          "Topics",
+          "Producer"
+        ],
+        "description": "Sends one or more records to a given topic, optionally specifying a partition, key, or both.",
+        "operationId": "send",
+        "consumes": [
+          "application/vnd.kafka.json.v2+json",
+          "application/vnd.kafka.binary.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/ProducerRecordList"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Records sent successfully.",
+            "schema": {
+              "$ref": "#/definitions/OffsetRecordSentList"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "offsets": [
+                  {
+                    "partition": 2,
+                    "offset": 0
+                  },
+                  {
+                    "partition": 1,
+                    "offset": 1
+                  },
+                  {
+                    "partition": 2,
+                    "offset": 2
+                  }
+                ]
+              }
+            }
+          },
+          "404": {
+            "description": "The specified topic was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified topic was not found."
+              }
+            }
+          },
+          "422": {
+            "description": "The record list is not valid.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 422,
+                "message": "The record list contains invalid records."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "topicname",
+          "in": "path",
+          "description": "Name of the topic to which you want to send records.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    },
+    "/consumers/{groupid}/instances/{name}/records": {
+      "get": {
+        "tags": [
+          "Consumers"
+        ],
+        "description": "Retrieves records for a subscribed consumer, including message values, topics, and partitions. The operation MUST be routed to the Kafka Bridge instance that contains the consumer.",
+        "operationId": "poll",
+        "produces": [
+          "application/vnd.kafka.json.v2+json",
+          "application/vnd.kafka.binary.v2+json",
+          "application/vnd.kafka.v2+json"
+        ],
+        "responses": {
+          "200": {
+            "description": "Poll request executed successfully.",
+            "schema": {
+              "$ref": "#/definitions/ConsumerRecordList"
+            },
+            "examples": {
+              "application/vnd.kafka.json.v2+json": [
+                {
+                  "topic": "topic",
+                  "key": "key1",
+                  "value": {
+                    "foo": "bar"
+                  },
+                  "partition": 0,
+                  "offset": 2
+                },
+                {
+                  "topic": "topic",
+                  "key": "key2",
+                  "value": [
+                    "foo2",
+                    "bar2"
+                  ],
+                  "partition": 1,
+                  "offset": 3
+                }
+              ],
+              "application/vnd.kafka.binary.v2+json": "[\n  {\n    \"topic\": \"test\",\n    \"key\": \"a2V5\",\n    \"value\": \"Y29uZmx1ZW50\",\n    \"partition\": 1,\n    \"offset\": 100,\n  },\n  {\n    \"topic\": \"test\",\n    \"key\": \"a2V5\",\n    \"value\": \"a2Fma2E=\",\n    \"partition\": 2,\n    \"offset\": 101,\n  }\n]"
+            }
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          },
+          "406": {
+            "description": "The `format` used in the consumer creation request does not match the embedded format in the Accept header of this request.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+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."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "groupid",
+          "in": "path",
+          "description": "ID of the consumer group to which the subscribed consumer belongs.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "name",
+          "in": "path",
+          "description": "Name of the subscribed consumer for which you want to retrieve records.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "timeout",
+          "in": "query",
+          "description": "The maximum amount of time, in milliseconds, that the HTTP Bridge spends retrieving records before timing out the request.",
+          "required": false,
+          "type": "integer"
+        },
+        {
+          "name": "max_bytes",
+          "in": "query",
+          "description": "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.",
+          "required": false,
+          "type": "integer"
+        }
+      ]
+    },
+    "/topics/{topicname}/partitions/{partitionid}": {
+      "post": {
+        "tags": [
+          "Topics",
+          "Producer"
+        ],
+        "description": "Sends one or more records to a given topic partition, optionally specifying a key.",
+        "operationId": "sendToPartition",
+        "consumes": [
+          "application/vnd.kafka.json.v2+json",
+          "application/vnd.kafka.binary.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "description": "List of records to send to a given topic partition, including a value (required) and a key (optional).",
+            "required": true,
+            "schema": {
+              "$ref": "#/definitions/ProducerRecordToPartitionList"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Records sent successfully.",
+            "schema": {
+              "$ref": "#/definitions/OffsetRecordSentList"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "offsets": [
+                  {
+                    "partition": 2,
+                    "offset": 0
+                  },
+                  {
+                    "partition": 1,
+                    "offset": 1
+                  },
+                  {
+                    "partition": 2,
+                    "offset": 2
+                  }
+                ]
+              }
+            }
+          },
+          "404": {
+            "description": "The specified topic partition was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified topic partition was not found."
+              }
+            }
+          },
+          "422": {
+            "description": "The record is not valid.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 422,
+                "message": "The record is not valid."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "topicname",
+          "in": "path",
+          "description": "Name of the topic containing the partition to which you want to send records.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "partitionid",
+          "in": "path",
+          "description": "ID of the partition to which you want to send records.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    },
+    "/consumers/{groupid}/instances/{name}": {
+      "delete": {
+        "tags": [
+          "Consumers"
+        ],
+        "description": "Deletes a specified consumer instance. The operation MUST be routed to the Kafka Bridge instance that contains the consumer.",
+        "operationId": "deleteConsumer",
+        "consumes": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "responses": {
+          "204": {
+            "description": "Consumer removed successfully."
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "groupid",
+          "in": "path",
+          "description": "ID of the consumer group to which the consumer belongs.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "name",
+          "in": "path",
+          "description": "Name of the consumer that you want to delete.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    },
+    "/consumers/{groupid}/instances/{name}/offsets": {
+      "post": {
+        "tags": [
+          "Consumers"
+        ],
+        "description": "Commits a list of consumer offsets. To commit offsets for all records fetched by the consumer, leave the request body empty.",
+        "operationId": "commit",
+        "consumes": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "parameters": [
+          {
+            "name": "body",
+            "in": "body",
+            "description": "List of consumer offsets to commit to the consumer offsets commit log. You can specify one or more topic partitions to commit offsets for.",
+            "required": false,
+            "schema": {
+              "$ref": "#/definitions/OffsetCommitSeekList"
+            }
+          }
+        ],
+        "responses": {
+          "204": {
+            "description": "Commit made successfully."
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          }
+        }
+      },
+      "parameters": [
+        {
+          "name": "groupid",
+          "in": "path",
+          "description": "ID of the consumer group to which the consumer belongs.",
+          "required": true,
+          "type": "string"
+        },
+        {
+          "name": "name",
+          "in": "path",
+          "description": "Name of the consumer.",
+          "required": true,
+          "type": "string"
+        }
+      ]
+    }
+  },
+  "definitions": {
+    "Consumer": {
+      "title": "Consumer",
+      "type": "object",
+      "properties": {
+        "name": {
+          "description": "The unique name for the consumer instance. The name is unique within the scope of the consumer group. The name is used in URLs. ",
+          "type": "string"
+        },
+        "format": {
+          "description": "The allowable message format for the consumer, which can be `binary` (default) or `json`. The messages are converted into a JSON format. ",
+          "type": "string"
+        },
+        "auto.offset.reset": {
+          "description": "Resets the offset position for the consumer. \nIf set to `earliest`, messages are read from the first offset. \nIf set to `latest`, messages are read from the latest offset.",
+          "type": "string"
+        },
+        "auto.commit.enable": {
+          "description": "If set to `true`, message offsets are committed automatically for the consumer. If set to `false`, message offsets must be committed manually.",
+          "type": "string"
+        },
+        "fetch.min.bytes": {
+          "description": "Sets the minimum ammount of data in bytes for the consumer to receive. The broker waits until the data to send exceeds this amount.",
+          "type": "string"
+        },
+        "consumer.request.timeout.ms": {
+          "description": "Sets the maximum amount of time, in milliseconds, for the consumer to wait for messages for a request. If the timeout period is reached without a response, an error is returned.",
+          "type": "string"
+        }
+      },
+      "example": {
+        "name": "consumer1",
+        "format": "binary",
+        "auto.offset.reset": "earliest",
+        "auto.commit.enable": "false",
+        "fetch.min.bytes": "512",
+        "consumer.request.timeout.ms": "30000"
+      }
+    },
+    "Partition": {
+      "title": "Partition",
+      "type": "object",
+      "properties": {
+        "partition": {
+          "format": "int32",
+          "type": "integer"
+        },
+        "topic": {
+          "type": "string"
+        }
+      }
+    },
+    "Topics": {
+      "title": "Topics",
+      "type": "object",
+      "properties": {
+        "topics": {
+          "type": "array",
+          "items": {
+            "type": "string"
+          }
+        }
+      },
+      "example": {
+        "topics": [
+          "topic1",
+          "topic2"
+        ]
+      }
+    },
+    "Partitions": {
+      "title": "Partitions",
+      "type": "object",
+      "properties": {
+        "partitions": {
+          "type": "array",
+          "items": {
+            "$ref": "#/definitions/Partition"
+          }
+        }
+      },
+      "example": {
+        "partitions": [
+          {
+            "topic": "topic",
+            "partition": 0
+          },
+          {
+            "topic": "topic",
+            "partition": 1
+          }
+        ]
+      }
+    },
+    "ConsumerRecordList": {
+      "title": "ConsumerRecordList",
+      "type": "array",
+      "items": {
+        "type": "object",
+        "properties": {
+          "topic": {
+            "type": "string"
+          },
+          "key": {
+            "type": "string"
+          },
+          "value": {
+            "type": "string"
+          },
+          "partition": {
+            "format": "int32",
+            "type": "integer"
+          },
+          "offset": {
+            "format": "int64",
+            "type": "long"
+          }
+        }
+      },
+      "example": [
+        {
+          "topic": "topic",
+          "key": "key1",
+          "value": "value1",
+          "partition": 0,
+          "offset": 2
+        },
+        {
+          "topic": "topic",
+          "key": "key2",
+          "value": "value2",
+          "partition": 1,
+          "offset": 3
+        }
+      ]
+    },
+    "CreatedConsumer": {
+      "title": "CreatedConsumer",
+      "type": "object",
+      "properties": {
+        "instance_id": {
+          "description": "Unique ID for the consumer instance in the group.",
+          "type": "string"
+        },
+        "base_uri": {
+          "description": "Base URI used to construct URIs for subsequent requests against this consumer instance.",
+          "type": "string"
+        }
+      },
+      "example": {
+        "instance_id": "my-consumer",
+        "base_uri": "http://localhost:8080/consumers/my-group/instances/my-consumer"
+      }
+    },
+    "OffsetCommitSeekList": {
+      "title": "OffsetCommitSeekList",
+      "type": "object",
+      "properties": {
+        "offsets": {
+          "type": "array",
+          "items": {
+            "$ref": "#/definitions/OffsetCommitSeek"
+          }
+        }
+      },
+      "example": {
+        "offsets": [
+          {
+            "topic": "topic",
+            "partition": 0,
+            "offset": 15
+          },
+          {
+            "topic": "topic",
+            "partition": 1,
+            "offset": 42
+          }
+        ]
+      }
+    },
+    "OffsetCommitSeek": {
+      "title": "OffsetCommitSeek",
+      "required": [],
+      "type": "object",
+      "properties": {
+        "partition": {
+          "format": "int32",
+          "type": "integer"
+        },
+        "offset": {
+          "format": "int64",
+          "type": "integer"
+        },
+        "topic": {
+          "type": "string"
+        }
+      },
+      "example": {
+        "partition": 43,
+        "offset": 92,
+        "topic": "topic"
+      }
+    },
+    "OffsetRecordSent": {
+      "title": "OffsetRecordSent",
+      "type": "object",
+      "properties": {
+        "partition": {
+          "format": "int32",
+          "type": "integer"
+        },
+        "offset": {
+          "format": "int64",
+          "type": "integer"
+        }
+      },
+      "example": {
+        "partition": 31,
+        "offset": 86
+      }
+    },
+    "OffsetRecordSentList": {
+      "title": "OffsetRecordSentList",
+      "type": "object",
+      "properties": {
+        "offsets": {
+          "type": "array",
+          "items": {
+            "$ref": "#/definitions/OffsetRecordSent"
+          }
+        }
+      },
+      "example": {
+        "offsets": [
+          {
+            "partition": 92,
+            "offset": 98
+          },
+          {
+            "partition": 65,
+            "offset": 91
+          }
+        ]
+      }
+    },
+    "Error": {
+      "title": "Error",
+      "type": "object",
+      "properties": {
+        "error_code": {
+          "format": "int32",
+          "type": "integer"
+        },
+        "message": {
+          "type": "string"
+        }
+      },
+      "example": "{\n    \"error_code\": 404,\n    \"message\": \"resource not found\"\n}"
+    },
+    "ProducerRecord": {
+      "title": "ProducerRecord",
+      "required": [
+        "value"
+      ],
+      "type": "object",
+      "properties": {
+        "partition": {
+          "format": "int32",
+          "type": "integer"
+        },
+        "value": {
+          "type": [
+            "object",
+            "string"
+          ]
+        },
+        "key": {
+          "type": [
+            "object",
+            "string"
+          ]
+        }
+      },
+      "example": "{\n    \"key\": \"key\"\n    \"partition\": 23\n    \"value\": {\"foo\": \"bar\"}\n}"
+    },
+    "ProducerRecordList": {
+      "title": "ProducerRecordList",
+      "type": "object",
+      "properties": {
+        "records": {
+          "type": "array",
+          "items": {
+            "$ref": "#/definitions/ProducerRecord"
+          }
+        }
+      },
+      "example": {
+        "records": [
+          {
+            "key": "key1",
+            "value": "value1"
+          },
+          {
+            "value": "value2",
+            "partition": 1
+          },
+          {
+            "value": "value3"
+          }
+        ]
+      }
+    },
+    "ProducerRecordToPartition": {
+      "title": "ProducerRecordToPartition",
+      "required": [
+        "value"
+      ],
+      "type": "object",
+      "properties": {
+        "value": {
+          "type": [
+            "object",
+            "string"
+          ]
+        },
+        "key": {
+          "type": [
+            "object",
+            "string"
+          ]
+        }
+      },
+      "example": {
+        "value": {
+          "v": 128
+        },
+        "key": {
+          "k": "key"
+        }
+      }
+    },
+    "ProducerRecordToPartitionList": {
+      "title": "ProducerRecordToPartitionList",
+      "type": "object",
+      "properties": {
+        "records": {
+          "type": "array",
+          "items": {
+            "$ref": "#/definitions/ProducerRecordToPartition"
+          }
+        }
+      },
+      "example": {
+        "records": [
+          {
+            "key": "key1",
+            "value": "value1"
+          },
+          {
+            "value": "value2"
+          }
+        ]
+      }
+    }
+  },
+  "tags": [
+    {
+      "name": "Consumers",
+      "description": "Consumer operations to create consumers in your Kafka cluster and perform common actions, such as subscribing to topics, retrieving processed records, and committing offsets."
+    },
+    {
+      "name": "Topics",
+      "description": "Topic operations to send messages to a specified topic or topic partition. You can include message keys in requests."
+    },
+    {
+      "name": "Seek",
+      "description": "Seek operations that enable a consumer to begin receiving messages from a given offset position."
+    },
+    {
+      "name": "Producer",
+      "description": "Producer operations to send records to a specified topic or topic partition."
+    }
+  ],
+  "externalDocs": {
+    "description": "Find out more about the Strimzi HTTP Bridge",
+    "url": "https://strimzi.io/documentation/"
+  }
+}
\ No newline at end of file


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
index fa161f3..d2d7e3a 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
@@ -414,7 +414,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + topic)
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
-                .putHeader("Content-Type", "application/vnd.kafka.json.v2+json")
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -478,6 +478,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -521,6 +522,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -547,6 +549,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -589,6 +592,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -629,6 +633,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -655,6 +660,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -697,6 +703,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -742,6 +749,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -769,6 +777,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -814,6 +823,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -859,6 +869,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -882,6 +893,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicPattern.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicPattern, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -927,6 +939,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -969,6 +982,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         // create a consumer
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -995,6 +1009,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/assignments")
                 .putHeader("Content-length", String.valueOf(partitionsRoot.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(partitionsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1037,7 +1052,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
-                .putHeader("Content-length", String.valueOf(0))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1077,10 +1092,12 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         JsonObject json = new JsonObject();
         json.put("name", name);
+        json.put("format", "json");
 
         // create a consumer
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1108,6 +1125,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/assignments")
                 .putHeader("Content-length", String.valueOf(partitionsRoot.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(partitionsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1121,6 +1139,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async consumeAsync = context.async();
 
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
+                .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1153,7 +1172,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
-                .putHeader("Content-length", String.valueOf(0))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1195,6 +1214,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1221,6 +1241,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1273,6 +1294,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/offsets")
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1280,7 +1302,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                     HttpResponse<JsonObject> response = ar.result();
 
                     int code = response.statusCode();
-                    context.assertEquals(200, code);
+                    context.assertEquals(204, code);
                     commitAsync.complete();
                 });
 
@@ -1290,6 +1312,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1331,6 +1354,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1357,6 +1381,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1404,7 +1429,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                     HttpResponse<Buffer> response = ar.result();
 
                     int code = response.statusCode();
-                    context.assertEquals(200, code);
+                    context.assertEquals(204, code);
                     commitAsync.complete();
                 });
 
@@ -1414,6 +1439,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1534,6 +1560,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + kafkaTopic)
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1582,6 +1609,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(subJson.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(subJson, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1735,6 +1763,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1753,6 +1782,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         // create the same consumer again
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1771,6 +1801,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         json.put("name", name + "diff");
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1801,6 +1832,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async consumeAsync = context.async();
 
         client.get(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/records" + "?timeout=" + String.valueOf(1000))
+                .putHeader("Accept", BridgeContentType.KAFKA_JSON_JSON)
                 .as(BodyCodec.jsonArray())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1837,6 +1869,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/offsets")
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1877,6 +1910,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1903,6 +1937,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1931,6 +1966,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1973,6 +2009,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -1995,6 +2032,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topics.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topics, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2027,6 +2065,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/positions")
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2080,6 +2119,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2117,6 +2157,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2140,6 +2181,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topics.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topics, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2176,6 +2218,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/positions/beginning")
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2205,6 +2248,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2237,6 +2281,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2260,6 +2305,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topics.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topics, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2299,6 +2345,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/positions/end")
                 .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2328,6 +2375,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2368,6 +2416,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2394,6 +2443,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2473,6 +2523,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2513,6 +2564,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2539,6 +2591,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2566,6 +2619,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2606,6 +2660,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, "/consumers/" + groupId)
                 .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(json, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2632,6 +2687,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         client.post(BRIDGE_PORT, BRIDGE_HOST, baseUri + "/subscription")
                 .putHeader("Content-length", String.valueOf(topicsRoot.toBuffer().length()))
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .sendJsonObject(topicsRoot, ar -> {
                     context.assertTrue(ar.succeeded());
@@ -2659,6 +2715,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         Async deleteAsync = context.async();
 
         client.delete(BRIDGE_PORT, BRIDGE_HOST, baseUri)
+                .putHeader("Content-Type", BridgeContentType.KAFKA_JSON)
                 .as(BodyCodec.jsonObject())
                 .send(ar -> {
                     context.assertTrue(ar.succeeded());