Repository: strimzi/strimzi-kafka-bridge
Pull request #315 (2019-09-03 09:19:54)

Scenarios for parameterized ST to create consumer
Added negative scenarios for parameterized ST to create consumer


Patch:
diff --git a/.travis/check_docs.sh b/.travis/check_docs.sh
index 1d602b9..b731b14 100755
--- a/.travis/check_docs.sh
+++ b/.travis/check_docs.sh
@@ -61,6 +61,6 @@ if [ -n "$CHANGED_DERIVED" ] ; then
   echo "Run the following to add up-to-date resources:"
   echo "  make docu_api \\"
   echo "    && git add documentation/book/ \\"
-  echo "    && git commit -m 'Update generated documentation'"
+  echo "    && git commit -s -m 'Update generated documentation'"
   exit 1
 fi
\ No newline at end of file
diff --git a/.travis/docu-push-to-website.sh b/.travis/docu-push-to-website.sh
index 5de22b2..a692d9b 100755
--- a/.travis/docu-push-to-website.sh
+++ b/.travis/docu-push-to-website.sh
@@ -24,7 +24,7 @@ git config user.name "Travis CI"
 git config user.email "ci@travis.tld"
 
 git add -A
-git commit -m "Update Kafka Bridge documentation (Travis CI build ${TRAVIS_BUILD_NUMBER})" --allow-empty
+git commit -s -m "Update Kafka Bridge documentation (Travis CI build ${TRAVIS_BUILD_NUMBER})" --allow-empty
 git push origin master
 
 popd
diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 0000000..8901434
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,18 @@
+# CHANGELOG
+
+## 0.14.0
+
+* Changed data types and enforced OpenAPI validation for `consumer.request.timeout.ms`, `enable.auto.commit`and `fetch.min.bytes`parameters on consumer creation. This is a breaking change.
+* Added HTTP GET method on `/consumers/{groupid}/instances/{name}/subscription` endpoint for getting subscribed topics and related assigned partitions.
+* Added automatic deletion of stale consumer after a configurable timeout if the HTTP DELETE is not called and the consumer is not used for long time.
+* Various bug fixes.
+
+## 0.13.0
+
+* Exposed "healthy" and "ready" endpoints as part of the OpenAPI specification
+* Added support for handling API gateway/proxy path routing to the bridge
+* Fixed OpenAPI v2 validation errors
+
+## 0.12.0
+
+* First release of the HTTP bridge
\ No newline at end of file
diff --git a/README.md b/README.md
index 32c3a16..8266550 100644
--- a/README.md
+++ b/README.md
@@ -1,4 +1,7 @@
 [![Build Status](https://travis-ci.org/strimzi/strimzi-kafka-bridge.svg?branch=master)](https://travis-ci.org/strimzi/strimzi-kafka-bridge)
+[![GitHub release](https://img.shields.io/github/release/strimzi/strimzi-kafka-bridge.svg)](https://github.com/strimzi/strimzi-kafka-bridge/releases/latest)
+[![License](https://img.shields.io/badge/license-Apache--2.0-blue.svg)](http://www.apache.org/licenses/LICENSE-2.0)
+[![Twitter Follow](https://img.shields.io/twitter/follow/strimziio.svg?style=social&label=Follow&style=for-the-badge)](https://twitter.com/strimziio)
 
 # Apache Kafka bridge
 
@@ -35,8 +38,8 @@ Documentation to the current _master_ branch as well as all releases can be foun
 
 If you encounter any issues while using Strimzi Kafka Bridge, you can get help using:
 
-- [Strimzi mailing list](https://www.redhat.com/mailman/listinfo/strimzi)
-- [Strimzi Slack workspace](https://join.slack.com/t/strimzi/shared_invite/enQtMzU2Mjk3NTgxMzE5LTYyMTUwMGNlMDQwMzBhOGI4YmY4MjhiMDgyNjA5OTk2MTFiYjc4M2Q3NGU1YTFjOWRiMzM2NGMwNDUwMjBlNDY)
+- [Strimzi Users mailing list](https://lists.cncf.io/g/cncf-strimzi-users/topics)
+- [#strimzi channel on CNCF Slack](https://slack.cncf.io/)
 
 ## Contributing
 
@@ -55,9 +58,8 @@ The [Documentation Contributor Guide](http://strimzi.io/contributing/guide/) des
 
 If you want to get in touch with us first before contributing, you can use:
 
-- [Strimzi mailing list](https://www.redhat.com/mailman/listinfo/strimzi)
-- [Strimzi Slack workspace](https://join.slack.com/t/strimzi/shared_invite/enQtMzU2Mjk3NTgxMzE5LTYyMTUwMGNlMDQwMzBhOGI4YmY4MjhiMDgyNjA5OTk2MTFiYjc4M2Q3NGU1YTFjOWRiMzM2NGMwNDUwMjBlNDY)
-
+- [Strimzi Dev mailing list](https://lists.cncf.io/g/cncf-strimzi-dev/topics)
+- [#strimzi channel on CNCF Slack](https://slack.cncf.io/)
 
 ## License
 
diff --git a/documentation/book/api/definitions.adoc b/documentation/book/api/definitions.adoc
index 77feda4..ef72fff 100644
--- a/documentation/book/api/definitions.adoc
+++ b/documentation/book/api/definitions.adoc
@@ -2,6 +2,11 @@
 [[_definitions]]
 == Definitions
 
+[[_assignedtopicpartitions]]
+=== AssignedTopicPartitions
+__Type__ : < string, < integer (int32) > array > map
+
+
 [[_consumer]]
 === Consumer
 
@@ -13,11 +18,11 @@ __optional__|Resets the offset position for the consumer.
 If set to `earliest`, messages are read from the first offset.
 If set to `latest`, messages are read from the latest offset.|string
 |**consumer.request.timeout.ms** +
-__optional__|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.|string
+__optional__|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.|integer
 |**enable.auto.commit** +
-__optional__|If set to `true`, message offsets are committed automatically for the consumer. If set to `false`, message offsets must be committed manually.|string
+__optional__|If set to `true`, message offsets are committed automatically for the consumer. If set to `false`, message offsets must be committed manually.|boolean
 |**fetch.min.bytes** +
-__optional__|Sets the minimum ammount of data, in bytes, for the consumer to receive. The broker waits until the data to send exceeds this amount.|string
+__optional__|Sets the minimum ammount of data, in bytes, for the consumer to receive. The broker waits until the data to send exceeds this amount.|integer
 |**format** +
 __optional__|The allowable message format for the consumer, which can be `binary` (default) or `json`. The messages are converted into a JSON format.|string
 |**name** +
@@ -25,12 +30,8 @@ __optional__|The unique name for the consumer instance. The name is unique withi
 |===
 
 
-[[_consumerrecordlist]]
-=== ConsumerRecordList
-__Type__ : < <<_consumerrecordlist_inline,ConsumerRecordList>> > array
-
-[[_consumerrecordlist_inline]]
-**ConsumerRecordList**
+[[_consumerrecord]]
+=== ConsumerRecord
 
 [options="header", cols=".^3,.^4"]
 |===
@@ -48,6 +49,11 @@ __optional__|string
 |===
 
 
+[[_consumerrecordlist]]
+=== ConsumerRecordList
+__Type__ : < <<_consumerrecord,ConsumerRecord>> > array
+
+
 [[_createdconsumer]]
 === CreatedConsumer
 
@@ -186,6 +192,19 @@ __optional__|< <<_producerrecordtopartition,ProducerRecordToPartition>> > array
 |===
 
 
+[[_subscribedtopiclist]]
+=== SubscribedTopicList
+
+[options="header", cols=".^3,.^4"]
+|===
+|Name|Schema
+|**partitions** +
+__optional__|< <<_assignedtopicpartitions,AssignedTopicPartitions>> > array
+|**topics** +
+__optional__|<<_topics,Topics>>
+|===
+
+
 [[_topics]]
 === Topics
 
diff --git a/documentation/book/api/paths.adoc b/documentation/book/api/paths.adoc
index d5e312d..3904d45 100644
--- a/documentation/book/api/paths.adoc
+++ b/documentation/book/api/paths.adoc
@@ -56,9 +56,9 @@ __required__|Name and configuration of the consumer. The name is unique within t
   "name" : "consumer1",
   "format" : "binary",
   "auto.offset.reset" : "earliest",
-  "enable.auto.commit" : "false",
-  "fetch.min.bytes" : "512",
-  "consumer.request.timeout.ms" : "30000"
+  "enable.auto.commit" : false,
+  "fetch.min.bytes" : 512,
+  "consumer.request.timeout.ms" : 30000
 }
 ----
 
@@ -770,6 +770,73 @@ __required__|List of topics to which the consumer will subscribe.|<<_topics,Topi
 ----
 
 
+[[_listsubscriptions]]
+=== GET /consumers/{groupid}/instances/{name}/subscription
+
+==== Description
+Retrieves a list of the topics to which the consumer is subscribed.
+
+
+==== Parameters
+
+[options="header", cols=".^2,.^3,.^9,.^4"]
+|===
+|Type|Name|Description|Schema
+|**Path**|**groupid** +
+__required__|ID of the consumer group to which the subscribed consumer belongs.|string
+|**Path**|**name** +
+__required__|Name of the consumer that you want to unsubscribe from topics.|string
+|===
+
+
+==== Responses
+
+[options="header", cols=".^2,.^14,.^4"]
+|===
+|HTTP Code|Description|Schema
+|**200**|List of subscribed topics and partitions.|<<_subscribedtopiclist,SubscribedTopicList>>
+|**404**|The specified consumer instance was not found.|<<_error,Error>>
+|===
+
+
+==== Produces
+
+* `application/vnd.kafka.v2+json`
+
+
+==== Tags
+
+* Consumers
+
+
+==== Example HTTP response
+
+===== Response 200
+[source,json]
+----
+{
+  "topics" : [ "my-topic1", "my-topic2" ],
+  "partitions" : [ {
+    "my-topic1" : [ 1, 2, 3 ]
+  }, {
+    "my-topic2" : [ 1 ]
+  } ]
+}
+----
+
+
+===== Response 404
+[source,json]
+----
+{
+  "application/vnd.kafka.v2+json" : {
+    "error_code" : 404,
+    "message" : "The specified consumer instance was not found."
+  }
+}
+----
+
+
 [[_unsubscribe]]
 === DELETE /consumers/{groupid}/instances/{name}/subscription
 
diff --git a/pom.xml b/pom.xml
index 6e3ccac..906c77d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -14,7 +14,7 @@
 		<maven.compiler.target>1.8</maven.compiler.target>
 		<slf4j-log4j12.version>1.7.18</slf4j-log4j12.version>
 		<log4j.version>1.2.17</log4j.version>
-		<vertx.version>3.8.0</vertx.version>
+		<vertx.version>3.8.1</vertx.version>
 		<debezium.version>0.8.3.Final</debezium.version>
 		<maven.checkstyle.version>3.0.0</maven.checkstyle.version>
 		<jupiter.version>5.4.0</jupiter.version>
diff --git a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
index 4832436..c92440b 100644
--- a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
@@ -194,6 +194,14 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
         this.consumer.unsubscribe(this::unsubscribeHandler);
     }
 
+    /**
+     * Returns all the topics which the consumer currently subscribes
+     */
+    protected void listSubscriptions(Handler<AsyncResult<Set<TopicPartition>>> handler) {
+        log.info("Listing subscribed topics {}", this.topicSubscriptions);
+        this.consumer.assignment(handler);
+    }
+
     /**
      * Subscribe to topics via the provided pattern represented by a Java regex
      *
@@ -573,7 +581,6 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
         this.unsubscribeHandler = handler;
     }
 
-
     /**
      * Set the handler called after a request for a specific partition is executed
      *
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
index df8965d..3f06211 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -128,6 +128,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
                 routerFactory.addHandlerByOperationId(this.DELETE_CONSUMER.getOperationId().toString(), this.DELETE_CONSUMER);
                 routerFactory.addHandlerByOperationId(this.SUBSCRIBE.getOperationId().toString(), this.SUBSCRIBE);
                 routerFactory.addHandlerByOperationId(this.UNSUBSCRIBE.getOperationId().toString(), this.UNSUBSCRIBE);
+                routerFactory.addHandlerByOperationId(this.LIST_SUBSCRIPTIONS.getOperationId().toString(), this.LIST_SUBSCRIPTIONS);
                 routerFactory.addHandlerByOperationId(this.ASSIGN.getOperationId().toString(), this.ASSIGN);
                 routerFactory.addHandlerByOperationId(this.POLL.getOperationId().toString(), this.POLL);
                 routerFactory.addHandlerByOperationId(this.COMMIT.getOperationId().toString(), this.COMMIT);
@@ -263,6 +264,11 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
         processConsumer(routingContext);
     }
 
+    private void listSubscriptions(RoutingContext routingContext) {
+        this.httpBridgeContext.setOpenApiOperation(HttpOpenApiOperations.LIST_SUBSCRIPTIONS);
+        processConsumer(routingContext);
+    }
+
     private void assign(RoutingContext routingContext) {
         this.httpBridgeContext.setOpenApiOperation(HttpOpenApiOperations.ASSIGN);
         processConsumer(routingContext);
@@ -506,6 +512,14 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
         }
     };
 
+    HttpOpenApiOperation LIST_SUBSCRIPTIONS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_SUBSCRIPTIONS) {
+
+        @Override
+        public void process(RoutingContext routingContext) {
+            listSubscriptions(routingContext);
+        }
+    };
+
     HttpOpenApiOperation ASSIGN = new HttpOpenApiOperation(HttpOpenApiOperations.ASSIGN) {
     
         @Override
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperations.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperations.java
index 9cca81f..54bdef2 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperations.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperations.java
@@ -16,6 +16,7 @@ public enum HttpOpenApiOperations {
     DELETE_CONSUMER("deleteConsumer"),
     SUBSCRIBE("subscribe"),
     UNSUBSCRIBE("unsubscribe"),
+    LIST_SUBSCRIPTIONS("listSubscriptions"),
     ASSIGN("assign"),
     POLL("poll"),
     COMMIT("commit"),
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 21c0c16..5ff6770 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -98,10 +98,17 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         Properties config = new Properties();
         addConfigParameter(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
             bodyAsJson.getString(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, null), config);
-        addConfigParameter(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
-            bodyAsJson.getString(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, null), config);
-        addConfigParameter(ConsumerConfig.FETCH_MIN_BYTES_CONFIG,
-            bodyAsJson.getString(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, null), config);
+        // OpenAPI validation handles boolean and integer, quoted or not as string, in the same way
+        // instead of raising a validation error due to this: https://github.com/vert-x3/vertx-web/issues/1375
+        Object enableAutoCommit = bodyAsJson.getValue(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
+        addConfigParameter(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, 
+            enableAutoCommit != null ? String.valueOf(enableAutoCommit) : null, config);
+        Object fetchMinBytes = bodyAsJson.getValue(ConsumerConfig.FETCH_MIN_BYTES_CONFIG);
+        addConfigParameter(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 
+            fetchMinBytes != null ? String.valueOf(fetchMinBytes) : null, config);
+        Object requestTimeoutMs = bodyAsJson.getValue("consumer." + ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG);
+        addConfigParameter(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,
+            requestTimeoutMs != null ? String.valueOf(requestTimeoutMs) : null, config);
         addConfigParameter(ConsumerConfig.CLIENT_ID_CONFIG, this.name, config);
 
         // create the consumer
@@ -345,6 +352,44 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         }
     }
 
+    public void doListSubscriptions(RoutingContext routingContext) {
+        this.listSubscriptions(listSubscriptionsResult -> {
+
+            if (listSubscriptionsResult.succeeded()) {
+                JsonObject root = new JsonObject();
+                JsonArray topicsArray = new JsonArray();
+                JsonArray partitionsArray = new JsonArray();
+
+                HashMap<String, JsonArray> partitions = new HashMap<>();
+                for (TopicPartition topicPartition: listSubscriptionsResult.result()) {
+                    if (!topicsArray.contains(topicPartition.getTopic())) {
+                        topicsArray.add(topicPartition.getTopic());
+                    }
+                    if (!partitions.containsKey(topicPartition.getTopic())) {
+                        partitions.put(topicPartition.getTopic(), new JsonArray());
+                    }
+                    partitions.put(topicPartition.getTopic(), partitions.get(topicPartition.getTopic()).add(topicPartition.getPartition()));
+                }
+                for (Map.Entry<String, JsonArray> part: partitions.entrySet()) {
+                    JsonObject topic = new JsonObject();
+                    topic.put(part.getKey(), part.getValue());
+                    partitionsArray.add(topic);
+                }
+                root.put("topics", topicsArray);
+                root.put("partitions", partitionsArray);
+
+                HttpUtils.sendResponse(routingContext, HttpResponseStatus.OK.code(), BridgeContentType.KAFKA_JSON, root.toBuffer());
+            } else {
+                HttpBridgeError error = new HttpBridgeError(
+                        HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                        listSubscriptionsResult.cause().getMessage()
+                );
+                HttpUtils.sendResponse(routingContext, HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                        BridgeContentType.KAFKA_JSON, error.toJson().toBuffer());
+            }
+        });
+    }
+
     public void doUnsubscribe(RoutingContext routingContext) {
         this.setUnsubscribeHandler(unsubscribeResult -> {
             if (unsubscribeResult.succeeded()) {
@@ -426,6 +471,9 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
             case UNSUBSCRIBE:
                 doUnsubscribe(routingContext);
                 break;
+            case LIST_SUBSCRIPTIONS:
+                doListSubscriptions(routingContext);
+                break;
 
             default:
                 throw new IllegalArgumentException("Unknown Operation: " + this.httpBridgeContext.getOpenApiOperation());
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
index 9988dad..510f606 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
@@ -26,6 +26,7 @@ import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.RoutingContext;
 import io.vertx.kafka.client.producer.KafkaProducerRecord;
 import io.vertx.kafka.client.producer.RecordMetadata;
+import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.serialization.Serializer;
 
 import java.util.ArrayList;
@@ -35,6 +36,7 @@ import java.util.UUID;
 public class HttpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
 
     private MessageConverter<K, V, Buffer, Buffer> messageConverter;
+    private boolean closing;
 
     public HttpSourceBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfig,
                                     EmbeddedFormat format, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
@@ -44,10 +46,12 @@ public class HttpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
     @Override
     public void open() {
         this.name = this.bridgeConfig.getBridgeID() == null ? "kafka-bridge-producer-" + UUID.randomUUID() : this.bridgeConfig.getBridgeID() + "-" + UUID.randomUUID();
+        this.closing = false;
         super.open();
     }
 
     @Override
+    @SuppressWarnings("checkstyle:NPathComplexity")
     public void handle(Endpoint<?> endpoint) {
         RoutingContext routingContext = (RoutingContext) endpoint.get();
 
@@ -95,19 +99,23 @@ public class HttpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
 
             for (int i = 0; i < sendHandlers.size(); i++) {
                 // check if, for each future, the sending operation is completed successfully or failed
-                if (done.result() != null && done.result().succeeded(i)) {
-                    RecordMetadata metadata = done.result().resultAt(i);
+                if (sendHandlers.get(i).succeeded() && sendHandlers.get(i).result() != null) {
+                    RecordMetadata metadata = (RecordMetadata) sendHandlers.get(i).result();
                     log.debug("Delivered record {} to Kafka on topic {} at partition {} [{}]", finalRecords.get(i), metadata.getTopic(), metadata.getPartition(), metadata.getOffset());
                     results.add(new HttpBridgeResult<>(metadata));
                 } else {
-                    String msg = done.cause().getMessage();
-                    int code = getCodeFromMsg(msg);
+                    String msg = sendHandlers.get(i).cause().getMessage();
+                    int code = handleError(sendHandlers.get(i).cause());
                     log.error("Failed to deliver record {}", finalRecords.get(i), done.cause());
                     results.add(new HttpBridgeResult<>(new HttpBridgeError(code, msg)));
                 }
             }
             HttpUtils.sendResponse(routingContext, HttpResponseStatus.OK.code(),
                     BridgeContentType.KAFKA_JSON, buildOffsets(results).toBuffer());
+            
+            if (this.closing) {
+                this.close();
+            }
         });
     }
 
@@ -137,8 +145,10 @@ public class HttpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
         return jsonResponse;
     }
 
-    private int getCodeFromMsg(String msg) {
-        if (msg.contains("Invalid partition")) {
+    private int handleError(Throwable ex) {
+        if (ex instanceof TimeoutException && ex.getMessage() != null &&
+            ex.getMessage().contains("not present in metadata")) {
+            this.closing = true;
             return HttpResponseStatus.NOT_FOUND.code();
         } else {
             return HttpResponseStatus.INTERNAL_SERVER_ERROR.code();
diff --git a/src/main/resources/openapi.json b/src/main/resources/openapi.json
index 51696d9..f52b77b 100644
--- a/src/main/resources/openapi.json
+++ b/src/main/resources/openapi.json
@@ -235,6 +235,43 @@
             ]
         },
         "/consumers/{groupid}/instances/{name}/subscription": {
+            "get": {
+                "tags": [
+                    "Consumers"
+                ],
+                "responses": {
+                    "200": {
+                        "description": "List of subscribed topics and partitions.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/SubscribedTopicList"
+                                }
+                            }
+                        }
+                    },
+                    "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."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                },
+                "operationId": "listSubscriptions",
+                "description": "Retrieves a list of the topics to which the consumer is subscribed."
+            },
             "post": {
                 "tags": [
                     "Consumers"
@@ -1066,30 +1103,41 @@
     },
     "components": {
         "schemas": {
+            "ConsumerRecord": {
+                "example": {
+                    "key": "key1",
+                    "offset": 2,
+                    "partition": 0,
+                    "topic": "topic",
+                    "value": "value1"
+                },
+                "properties": {
+                    "key": {
+                        "type": "string"
+                    },
+                    "offset": {
+                        "format": "int64",
+                        "type": "integer"
+                    },
+                    "partition": {
+                        "format": "int32",
+                        "type": "integer"
+                    },
+                    "topic": {
+                        "type": "string"
+                    },
+                    "value": {
+                        "type": "string"
+                    }
+                },
+                "title": "ConsumerRecord",
+                "type": "object"
+            },
             "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": "integer"
-                        }
-                    }
+                    "$ref": "#/components/schemas//ConsumerRecord"
                 },
                 "example": [
                     {
@@ -1200,15 +1248,15 @@
                     },
                     "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"
+                        "type": "integer"
                     },
                     "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"
+                        "type": "integer"
                     },
                     "enable.auto.commit": {
                         "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"
+                        "type": "boolean"
                     }
                 },
                 "additionalProperties": false,
@@ -1216,9 +1264,9 @@
                     "name": "consumer1",
                     "format": "binary",
                     "auto.offset.reset": "earliest",
-                    "enable.auto.commit": "false",
-                    "fetch.min.bytes": "512",
-                    "consumer.request.timeout.ms": "30000"
+                    "enable.auto.commit": false,
+                    "fetch.min.bytes": 512,
+                    "consumer.request.timeout.ms": 30000
                 }
             },
             "OffsetCommitSeek": {
@@ -1462,6 +1510,59 @@
                         "topic2"
                     ]
                 }
+            },
+            "AssignedTopicPartitions": {
+                "title": "AssignedTopicPartitions",
+                "type": "object",
+                "additionalProperties": {
+                    "type": "array",
+                    "items": {
+                        "format": "int32",
+                        "type": "integer"
+                    }
+                },
+                "example": {
+                    "my-topic1": [
+                        1,
+                        2,
+                        3
+                    ]
+                }
+            },
+            "SubscribedTopicList": {
+                "title": "SubscribedTopicList",
+                "type": "object",
+                "properties": {
+                    "topics": {
+                        "$ref": "#/components/schemas/Topics"
+                    },
+                    "partitions": {
+                        "type": "array",
+                        "items": {
+                            "$ref": "#/components/schemas/AssignedTopicPartitions"
+                        }
+                    }
+                },
+                "example": {
+                    "topics": [
+                        "my-topic1",
+                        "my-topic2"
+                    ],
+                    "partitions": [
+                        {
+                            "my-topic1": [
+                                1,
+                                2,
+                                3
+                            ]
+                        },
+                        {
+                            "my-topic2": [
+                                1
+                            ]
+                        }
+                    ]
+                }
             }
         }
     },
diff --git a/src/main/resources/openapiv2.json b/src/main/resources/openapiv2.json
index 1aef287..56fc455 100644
--- a/src/main/resources/openapiv2.json
+++ b/src/main/resources/openapiv2.json
@@ -219,6 +219,36 @@
       ]
     },
     "/consumers/{groupid}/instances/{name}/subscription": {
+      "get": {
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "tags": [
+          "Consumers"
+        ],
+        "responses": {
+          "200": {
+            "description": "List of subscribed topics and partitions.",
+            "schema": {
+              "$ref": "#/definitions/SubscribedTopicList"
+            }
+          },
+          "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."
+              }
+            }
+          }
+        },
+        "operationId": "listSubscriptions",
+        "description": "Retrieves a list of the topics to which the consumer is subscribed."
+      },
       "post": {
         "tags": [
           "Consumers"
@@ -896,30 +926,41 @@
     }
   },
   "definitions": {
+    "ConsumerRecord": {
+      "example": {
+        "key": "key1",
+        "offset": 2,
+        "partition": 0,
+        "topic": "topic",
+        "value": "value1"
+      },
+      "properties": {
+        "key": {
+          "type": "string"
+        },
+        "offset": {
+          "format": "int64",
+          "type": "integer"
+        },
+        "partition": {
+          "format": "int32",
+          "type": "integer"
+        },
+        "topic": {
+          "type": "string"
+        },
+        "value": {
+          "type": "string"
+        }
+      },
+      "title": "ConsumerRecord",
+      "type": "object"
+    },
     "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": "integer"
-          }
-        }
+        "$ref": "#/definitions/ConsumerRecord"
       },
       "example": [
         {
@@ -1030,15 +1071,15 @@
         },
         "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"
+          "type": "integer"
         },
         "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"
+          "type": "integer"
         },
         "enable.auto.commit": {
           "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"
+          "type": "boolean"
         }
       },
       "additionalProperties": false,
@@ -1046,9 +1087,9 @@
         "name": "consumer1",
         "format": "binary",
         "auto.offset.reset": "earliest",
-        "enable.auto.commit": "false",
-        "fetch.min.bytes": "512",
-        "consumer.request.timeout.ms": "30000"
+        "enable.auto.commit": false,
+        "fetch.min.bytes": 512,
+        "consumer.request.timeout.ms": 30000
       }
     },
     "OffsetCommitSeek": {
@@ -1276,6 +1317,59 @@
           "topic2"
         ]
       }
+    },
+    "AssignedTopicPartitions": {
+      "title": "AssignedTopicPartitions",
+      "type": "object",
+      "additionalProperties": {
+        "type": "array",
+        "items": {
+          "format": "int32",
+          "type": "integer"
+        }
+      },
+      "example": {
+        "my-topic1": [
+          1,
+          2,
+          3
+        ]
+      }
+    },
+    "SubscribedTopicList": {
+      "title": "SubscribedTopicList",
+      "type": "object",
+      "properties": {
+        "topics": {
+          "$ref": "#/definitions/Topics"
+        },
+        "partitions": {
+          "type": "array",
+          "items": {
+            "$ref": "#/definitions/AssignedTopicPartitions"
+          }
+        }
+      },
+      "example": {
+        "topics": [
+          "my-topic1",
+          "my-topic2"
+        ],
+        "partitions": [
+          {
+            "my-topic1": [
+              1,
+              2,
+              3
+            ]
+          },
+          {
+            "my-topic2": [
+              1
+            ]
+          }
+        ]
+      }
     }
   },
   "tags": [


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionTest.java b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionTest.java
index 27e8ad1..d419346 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionTest.java
@@ -5,6 +5,7 @@
 package io.strimzi.kafka.bridge.http;
 
 import io.netty.handler.codec.http.HttpResponseStatus;
+import io.strimzi.kafka.bridge.BridgeContentType;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
 import io.strimzi.kafka.bridge.utils.Urls;
 import io.vertx.core.json.JsonArray;
@@ -156,4 +157,62 @@ public class ConsumerSubscriptionTest extends HttpBridgeTestBase {
         subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
         context.completeNow();
     }
+
+    @Test
+    void listConsumerSubscriptions(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
+        String topic = "listConsumerSubscriptions";
+        String topic2 = "listConsumerSubscriptions2";
+        kafkaCluster.createTopic(topic, 1, 1);
+        kafkaCluster.createTopic(topic2, 4, 1);
+
+        String name = "my-kafka-consumer";
+        String groupId = "my-group";
+
+        JsonObject json = new JsonObject();
+        json.put("name", name);
+        json.put("format", "json");
+
+        JsonArray topics = new JsonArray();
+        topics.add(topic);
+        topics.add(topic2);
+
+        JsonObject topicsRoot = new JsonObject();
+        topicsRoot.put("topics", topics);
+
+        consumerService()
+                .createConsumer(context, groupId, json)
+                .subscribeConsumer(context, groupId, name, topic)
+                .subscribeConsumer(context, groupId, name, topic2);
+
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
+
+        // poll to subscribe
+        consumerService()
+                .consumeRecordsRequest(groupId, name, BridgeContentType.KAFKA_JSON_JSON)
+                .as(BodyCodec.jsonObject())
+                .send(ar -> consume.complete(true));
+
+        CompletableFuture<Boolean> listSubscriptions = new CompletableFuture<>();
+        consumerService()
+                .listSubscriptionsConsumerRequest(groupId, name)
+                .as(BodyCodec.jsonObject())
+                .send(ar -> {
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.OK.code(), response.statusCode());
+                        assertEquals(response.body().getJsonArray("topics").size(), 2);
+                        assertTrue(response.body().getJsonArray("topics").contains(topic));
+                        assertTrue(response.body().getJsonArray("topics").contains(topic2));
+                        assertEquals(response.body().getJsonArray("partitions").size(), 2);
+                        assertEquals(response.body().getJsonArray("partitions").getJsonObject(0).getJsonArray(topic2).size(), 4);
+                        assertEquals(response.body().getJsonArray("partitions").getJsonObject(1).getJsonArray(topic).size(), 1);
+                        listSubscriptions.complete(true);
+                        context.completeNow();
+                    });
+                });
+
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
+    }
 }
\ No newline at end of file
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerTest.java b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerTest.java
index 743302f..0c59695 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerTest.java
@@ -54,6 +54,105 @@ public class ConsumerTest extends HttpBridgeTestBase {
         assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
+    @Test
+    void createConsumerEnableAutoCommit(VertxTestContext context) throws InterruptedException, TimeoutException, ExecutionException {
+        // both "true" ("false") and true (false) works due to https://github.com/vert-x3/vertx-web/issues/1375
+        // and the current OpenAPI validation doesn't recognize the first one as invalid
+        
+        JsonObject consumerJson = new JsonObject()
+            .put("name", name)
+            .put("enable.auto.commit", true);
+
+        // create consumer
+        consumerService().createConsumer(context, groupId, consumerJson);
+
+        consumerJson
+            .put("name", name + "-1")
+            .put("enable.auto.commit", "true");
+
+        // create consumer
+        consumerService().createConsumer(context, groupId, consumerJson);
+
+        consumerJson
+            .put("name", name + "-2")
+            .put("enable.auto.commit", "foo");
+
+        // create consumer
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
+        consumerService().createConsumerRequest(groupId, consumerJson)
+                .sendJsonObject(consumerJson, ar -> {
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.BAD_REQUEST.code(), response.statusCode());
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.BAD_REQUEST.code(), error.getCode());
+                        assertEquals("Validation error on: body.enable.auto.commit - $.enable.auto.commit: string found, boolean expected",
+                                error.getMessage());
+                    });
+                    create.complete(true);
+                });
+
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
+    }
+
+    @Test
+    void createConsumerFetchMinBytes(VertxTestContext context) throws InterruptedException, TimeoutException, ExecutionException {
+        this.createConsumerIntegerParam(context, "fetch.min.bytes");
+    }
+
+    @Test
+    void createConsumerRequestTimeoutMs(VertxTestContext context) throws InterruptedException, TimeoutException, ExecutionException {
+        this.createConsumerIntegerParam(context, "consumer.request.timeout.ms");
+    }
+
+    private void createConsumerIntegerParam(VertxTestContext context, String param) throws InterruptedException, TimeoutException, ExecutionException {
+        // both "100" and 100 works due to https://github.com/vert-x3/vertx-web/issues/1375
+        // and the current OpenAPI validation doesn't recognize the first one as invalid
+        
+        JsonObject consumerJson = new JsonObject()
+            .put("name", name)
+            .put(param, 100);
+
+        // create consumer
+        consumerService().createConsumer(context, groupId, consumerJson);
+
+        consumerJson
+            .put("name", name + "-1")
+            .put(param, "100");
+
+        // create consumer
+        consumerService().createConsumer(context, groupId, consumerJson);
+
+        consumerJson
+            .put("name", name + "-2")
+            .put(param, "foo");
+
+        // create consumer
+        CompletableFuture<Boolean> create = new CompletableFuture<>();
+        consumerService().createConsumerRequest(groupId, consumerJson)
+                .sendJsonObject(consumerJson, ar -> {
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.BAD_REQUEST.code(), response.statusCode());
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertEquals(HttpResponseStatus.BAD_REQUEST.code(), error.getCode());
+                        assertEquals("Validation error on: body." + param + " - $." + param + ": string found, integer expected",
+                                error.getMessage());
+                    });
+                    create.complete(true);
+                });
+
+        create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
+    }
+
     @Test
     void createConsumerWithForwardedHeaders(VertxTestContext context) throws InterruptedException {
         // this test emulates a create consumer request coming from an API gateway/proxy
@@ -185,25 +284,38 @@ public class ConsumerTest extends HttpBridgeTestBase {
     }
 
     @Test
-    void createConsumerWithWrongParameter(VertxTestContext context) throws InterruptedException {
-        JsonObject json = new JsonObject();
-        json.put("name", name);
-        json.put("auto.offset.reset", "foo");
+    void createConsumerWithWrongAutoOffsetReset(VertxTestContext context) throws InterruptedException, TimeoutException, ExecutionException {
+        checkCreatingConsumer("auto.offset.reset", "foo", HttpResponseStatus.INTERNAL_SERVER_ERROR,
+                "Invalid value foo for configuration auto.offset.reset: String must be one of: latest, earliest, none", context);
 
-        consumerService().createConsumerRequest(groupId, json)
-                .sendJsonObject(json, ar -> {
-                    context.verify(() -> {
-                        assertTrue(ar.succeeded());
-                        HttpResponse<JsonObject> response = ar.result();
-                        assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), response.statusCode());
-                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
-                        assertEquals(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), error.getCode());
-                        assertEquals("Invalid value foo for configuration auto.offset.reset: String must be one of: latest, earliest, none",
-                                error.getMessage());
-                    });
-                    context.completeNow();
-                });
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
+    }
 
+    @Test
+    void createConsumerWithWrongEnableAutoCommit(VertxTestContext context) throws InterruptedException, TimeoutException, ExecutionException {
+        checkCreatingConsumer("enable.auto.commit", "foo", HttpResponseStatus.INTERNAL_SERVER_ERROR,
+                "Invalid value foo for configuration enable.auto.commit: Expected value to be either true or false", context);
+
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
+    }
+
+    @Test
+    void createConsumerWithWrongFetchMinBytes(VertxTestContext context) throws InterruptedException, TimeoutException, ExecutionException {
+        checkCreatingConsumer("fetch.min.bytes", "foo", HttpResponseStatus.INTERNAL_SERVER_ERROR,
+                "Invalid value foo for configuration fetch.min.bytes: Not a number of type INT", context);
+
+        context.completeNow();
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
+    }
+
+    @Test
+    void createConsumerWithNotExistingParameter(VertxTestContext context) throws InterruptedException, TimeoutException, ExecutionException {
+        checkCreatingConsumer("foo", "bar", HttpResponseStatus.BAD_REQUEST,
+                "Validation error on: body - $.foo: is not defined in the schema and the schema does not allow additional properties", context);
+
+        context.completeNow();
         assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
     }
 
@@ -1167,4 +1279,25 @@ public class ConsumerTest extends HttpBridgeTestBase {
                     create.complete(true);
                 });
     }
+
+    private void checkCreatingConsumer(String key, String value, HttpResponseStatus status, String message, VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
+        JsonObject json = new JsonObject();
+        json.put("name", name);
+        json.put(key, value);
+
+        CompletableFuture<Boolean> consumer = new CompletableFuture<>();
+        consumerService().createConsumerRequest(groupId, json)
+            .sendJsonObject(json, ar -> {
+                context.verify(() -> {
+                    assertTrue(ar.succeeded());
+                    HttpResponse<JsonObject> response = ar.result();
+                    assertEquals(status.code(), response.statusCode());
+                    HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                    assertEquals(status.code(), error.getCode());
+                    assertEquals(message, error.getMessage());
+                });
+                consumer.complete(true);
+            });
+        consumer.get(TEST_TIMEOUT, TimeUnit.SECONDS);
+    }
 }
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTestBase.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTestBase.java
index 2108cc5..fd1c807 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTestBase.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTestBase.java
@@ -9,6 +9,7 @@ import io.strimzi.kafka.bridge.HealthChecker;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
 import io.strimzi.kafka.bridge.config.KafkaConsumerConfig;
+import io.strimzi.kafka.bridge.config.KafkaProducerConfig;
 import io.strimzi.kafka.bridge.facades.KafkaFacade;
 import io.strimzi.kafka.bridge.http.services.BaseService;
 import io.strimzi.kafka.bridge.http.services.ConsumerService;
@@ -21,6 +22,7 @@ import io.vertx.ext.web.client.WebClientOptions;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
@@ -39,6 +41,7 @@ class HttpBridgeTestBase {
     static {
         config.put(KafkaConfig.KAFKA_CONFIG_PREFIX + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
         config.put(KafkaConsumerConfig.KAFKA_CONSUMER_CONFIG_PREFIX + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        config.put(KafkaProducerConfig.KAFKA_PRODUCER_CONFIG_PREFIX + ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000");
         config.put(HttpConfig.HTTP_CONSUMER_TIMEOUT, timeout);
         config.put(BridgeConfig.BRIDGE_ID, "my-bridge");
     }
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesTest.java b/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesTest.java
index 92715e0..f8446ae 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesTest.java
@@ -73,8 +73,8 @@ public class OtherServicesTest extends HttpBridgeTestBase {
                         JsonObject bridgeResponse = response.body();
 
                         Map<String, Object> paths = bridgeResponse.getJsonObject("paths").getMap();
-                        // subscribe and unsubscribe are using the same endpoint but different methods
-                        int pathsSize = HttpOpenApiOperations.values().length - 1;
+                        // subscribe, list subscribtions and unsubscribe are using the same endpoint but different methods
+                        int pathsSize = HttpOpenApiOperations.values().length - 2;
                         assertEquals(pathsSize, paths.size());
                         assertTrue(paths.containsKey("/consumers/{groupid}"));
                         assertEquals(HttpOpenApiOperations.CREATE_CONSUMER.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}").getJsonObject("post").getString("operationId"));
@@ -87,6 +87,7 @@ public class OtherServicesTest extends HttpBridgeTestBase {
                         assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/subscription"));
                         assertEquals(HttpOpenApiOperations.SUBSCRIBE.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/subscription").getJsonObject("post").getString("operationId"));
                         assertEquals(HttpOpenApiOperations.UNSUBSCRIBE.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/subscription").getJsonObject("delete").getString("operationId"));
+                        assertEquals(HttpOpenApiOperations.LIST_SUBSCRIPTIONS.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/subscription").getJsonObject("get").getString("operationId"));
                         assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/assignments"));
                         assertEquals(HttpOpenApiOperations.ASSIGN.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/assignments").getJsonObject("post").getString("operationId"));
                         assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/records"));
@@ -104,7 +105,7 @@ public class OtherServicesTest extends HttpBridgeTestBase {
                         assertTrue(paths.containsKey("/openapi"));
                         assertEquals(HttpOpenApiOperations.OPENAPI.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/openapi").getJsonObject("get").getString("operationId"));
                         assertFalse(paths.containsKey("/karel"));
-                        assertEquals(15, bridgeResponse.getJsonObject("definitions").getMap().size());
+                        assertEquals(18, bridgeResponse.getJsonObject("definitions").getMap().size());
                         assertEquals(4, bridgeResponse.getJsonArray("tags").size());
                     });
                     context.completeNow();
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ProducerTest.java b/src/test/java/io/strimzi/kafka/bridge/http/ProducerTest.java
index 88ae666..32e0aa6 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ProducerTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ProducerTest.java
@@ -6,6 +6,7 @@ package io.strimzi.kafka.bridge.http;
 
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.strimzi.kafka.bridge.BridgeContentType;
+import io.strimzi.kafka.bridge.config.KafkaProducerConfig;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
 import io.strimzi.kafka.bridge.utils.KafkaJsonDeserializer;
 import io.vertx.core.AsyncResult;
@@ -16,9 +17,9 @@ import io.vertx.ext.web.client.HttpResponse;
 import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.consumer.KafkaConsumer;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -56,7 +57,7 @@ public class ProducerTest extends HttpBridgeTestBase {
 
         Properties config = kafkaCluster.getConsumerProperties();
 
-        KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config,
+        KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config,
                 new StringDeserializer(), new KafkaJsonDeserializer<>(String.class));
         consumer.handler(record -> {
             context.verify(() -> {
@@ -103,7 +104,7 @@ public class ProducerTest extends HttpBridgeTestBase {
 
         Properties config = kafkaCluster.getConsumerProperties();
 
-        KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config,
+        KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config,
                 new StringDeserializer(), new KafkaJsonDeserializer<>(String.class));
         consumer.handler(record -> {
             context.verify(() -> {
@@ -149,7 +150,7 @@ public class ProducerTest extends HttpBridgeTestBase {
 
         Properties config = kafkaCluster.getConsumerProperties();
 
-        KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config,
+        KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config,
                 new KafkaJsonDeserializer<>(String.class), new KafkaJsonDeserializer<>(String.class));
         consumer.handler(record -> {
             context.verify(() -> {
@@ -195,7 +196,7 @@ public class ProducerTest extends HttpBridgeTestBase {
 
         Properties config = kafkaCluster.getConsumerProperties();
 
-        KafkaConsumer<byte[], byte[]> consumer = KafkaConsumer.create(this.vertx, config,
+        KafkaConsumer<byte[], byte[]> consumer = KafkaConsumer.create(vertx, config,
                 new ByteArrayDeserializer(), new ByteArrayDeserializer());
         consumer.handler(record -> {
             context.verify(() -> {
@@ -226,12 +227,12 @@ public class ProducerTest extends HttpBridgeTestBase {
 
         Properties config = kafkaCluster.getConsumerProperties();
 
-        KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config,
+        KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config,
                 new KafkaJsonDeserializer<>(String.class), new KafkaJsonDeserializer<>(String.class));
 
         this.count = 0;
 
-        this.vertx.setPeriodic(HttpBridgeTestBase.PERIODIC_DELAY, timerId -> {
+        vertx.setPeriodic(HttpBridgeTestBase.PERIODIC_DELAY, timerId -> {
 
             if (this.count < HttpBridgeTestBase.PERIODIC_MAX_MESSAGE) {
 
@@ -250,7 +251,7 @@ public class ProducerTest extends HttpBridgeTestBase {
 
                 this.count++;
             } else {
-                this.vertx.cancelTimer(timerId);
+                vertx.cancelTimer(timerId);
 
                 consumer.subscribe(topic, done -> {
                     if (!done.succeeded()) {
@@ -320,7 +321,7 @@ public class ProducerTest extends HttpBridgeTestBase {
 
         Properties config = kafkaCluster.getConsumerProperties();
 
-        KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config,
+        KafkaConsumer<String, String> consumer = KafkaConsumer.create(vertx, config,
                 new StringDeserializer(), new KafkaJsonDeserializer<String>(String.class));
         this.count = 0;
         consumer.handler(record -> {
@@ -399,13 +400,17 @@ public class ProducerTest extends HttpBridgeTestBase {
 
                     HttpBridgeError error = HttpBridgeError.fromJson(offsets.getJsonObject(0));
                     assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
-                    assertEquals("Invalid partition given with record: 1000 is not in the range [0...3).", error.getMessage());
+                    // the message got from the Kafka producer (starting from 2.3) is misleading
+                    // this JIRA (https://issues.apache.org/jira/browse/KAFKA-8862) raises the issue
+                    assertEquals(
+                            "Topic " + kafkaTopic + " not present in metadata after " +
+                                    config.get(KafkaProducerConfig.KAFKA_PRODUCER_CONFIG_PREFIX + ProducerConfig.MAX_BLOCK_MS_CONFIG) + " ms.",
+                            error.getMessage());
                 });
                 context.completeNow();
             });
     }
 
-    @Disabled
     @Test
     void sendToNonExistingTopicTest(VertxTestContext context) {
         String kafkaTopic = "sendToNonExistingTopicTest";
@@ -433,10 +438,12 @@ public class ProducerTest extends HttpBridgeTestBase {
                     JsonArray offsets = bridgeResponse.getJsonArray("offsets");
                     assertEquals(1, offsets.size());
                     int code = offsets.getJsonObject(0).getInteger("error_code");
-                    String statusMessage = offsets.getJsonObject(0).getString("error");
+                    String statusMessage = offsets.getJsonObject(0).getString("message");
 
                     assertEquals(HttpResponseStatus.NOT_FOUND.code(), code);
-                    assertEquals("Topic " + kafkaTopic + " not found", statusMessage);
+                    assertEquals("Topic " + kafkaTopic + " not present in metadata after " + 
+                                config.get(KafkaProducerConfig.KAFKA_PRODUCER_CONFIG_PREFIX + ProducerConfig.MAX_BLOCK_MS_CONFIG) + " ms.", 
+                                statusMessage);
                 });
                 context.completeNow();
             });
@@ -592,4 +599,49 @@ public class ProducerTest extends HttpBridgeTestBase {
                 assertEquals(0L, metadata.getLong("offset"));
             });
     }
+
+    @Test
+    void sendMultipleRecordsWithOneInvalidPartitionTest(VertxTestContext context) {
+        String kafkaTopic = "sendMultipleRecordsWithOneInvalidPartitionTest";
+        kafkaCluster.createTopic(kafkaTopic, 3, 1);
+
+        String value = "Hi, This is kafka bridge";
+        int partition = 1;
+
+        JsonArray records = new JsonArray();
+        JsonObject json = new JsonObject();
+        json.put("value", value);
+        json.put("partition", partition);
+        records.add(json);
+
+        JsonObject json2 = new JsonObject();
+        json2.put("value", value + "invalid");
+        json2.put("partition", 500);
+        records.add(json2);
+
+        JsonObject root = new JsonObject();
+        root.put("records", records);
+
+        producerService()
+                .sendRecordsRequest(kafkaTopic, root, BridgeContentType.KAFKA_JSON_JSON)
+                .sendJsonObject(root, ar -> {
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.OK.code(), response.statusCode());
+                        JsonObject bridgeResponse = response.body();
+
+                        JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                        assertEquals(2, offsets.size());
+                        JsonObject metadata = offsets.getJsonObject(0);
+                        assertNotNull(metadata.getInteger("partition"));
+                        assertEquals(partition, metadata.getInteger("partition"));
+                        assertEquals(0L, metadata.getLong("offset"));
+                        HttpBridgeError error = HttpBridgeError.fromJson(offsets.getJsonObject(1));
+                        assertEquals(HttpResponseStatus.NOT_FOUND.code(), error.getCode());
+                        assertEquals("Topic " + kafkaTopic + " not present in metadata after 10000 ms.", error.getMessage());
+                    });
+                    context.completeNow();
+                });
+    }
 }
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/services/BaseService.java b/src/test/java/io/strimzi/kafka/bridge/http/services/BaseService.java
index 90a957a..533407e 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/services/BaseService.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/services/BaseService.java
@@ -18,7 +18,7 @@ public class BaseService {
     static final int HTTP_REQUEST_TIMEOUT = 60;
 
     // for request configuration
-    private static final long RESPONSE_TIMEOUT = 2000L;
+    private static final long RESPONSE_TIMEOUT = 60000L;
 
     BaseService(WebClient webClient) {
         this.webClient = webClient;
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java b/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java
index 17f3c20..1130a89 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java
@@ -71,6 +71,11 @@ public class ConsumerService extends BaseService {
                 .putHeader(CONTENT_TYPE.toString(), BridgeContentType.KAFKA_JSON);
     }
 
+    public HttpRequest<Buffer> listSubscriptionsConsumerRequest(String groupId, String name) {
+        return getRequest(Urls.consumerInstanceSubscription(groupId, name))
+                .putHeader(ACCEPT.toString(), BridgeContentType.KAFKA_JSON);
+    }
+
     public HttpRequest<JsonObject> offsetsRequest(String groupId, String name, JsonObject json) {
         return postRequest(Urls.consumerInstanceOffsets(groupId, name))
                 .putHeader(CONTENT_LENGTH.toString(), String.valueOf(json.toBuffer().length()))