Repository: strimzi/strimzi-kafka-bridge (java)

Issue:
Records are not saved if at least one of them has validation error
Kafka Bridge API returns error messages for all records which you want to save if at list one of them has an issue with validation. For example, if try to save 3 records and one of them has wrong `partition`, Bridge API will return an error for all records, even for those which are valid.

**Step to reproduce:**
1. Deploy Kafka Bridge
2. Create a topic with the name "test"
3. Send a POST request to the endpoint `/topics/{topicname}` with the following JSON:
```
{
  "records": [
    {
      "key": "key1",
      "value": "value1"
    },
    {
      "value": "value2",
      "partition": 1 <------- wrong value (should be 0)
    },
    {
      "value": "value3"
    }
  ]
}
```
Actual result:
Response has 3 errors for all messages ```
{
    "offsets": [
        {
            "error_code": 404,
            "message": "Invalid partition given with record: 1 is not in the range [0...1)."
        },
        {
            "error_code": 404,
            "message": "Invalid partition given with record: 1 is not in the range [0...1)."
        },
        {
            "error_code": 404,
            "message": "Invalid partition given with record: 1 is not in the range [0...1)."
        }
    ]
}
```
but actually we have only one wrong record with ```"partition": 1```
Expected result:
Record 1, 3 should be saved and only record 2 should have error.


Pull request:
Multi records partition bug
Fixes #310 and also refers to #312 

Patch:
diff --git a/documentation/book/api/definitions.adoc b/documentation/book/api/definitions.adoc
index 77feda4..b2ea0b2 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
 
@@ -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..6fba4db 100644
--- a/documentation/book/api/paths.adoc
+++ b/documentation/book/api/paths.adoc
@@ -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/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..5feef51 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -345,6 +345,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 +464,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 ecf86a8..da5d69e 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
@@ -96,13 +96,13 @@ 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 = getErrorCode(done.cause());
+                    String msg = sendHandlers.get(i).cause().getMessage();
+                    int code = getErrorCode(sendHandlers.get(i).cause());
                     log.error("Failed to deliver record {}", finalRecords.get(i), done.cause());
                     results.add(new HttpBridgeResult<>(new HttpBridgeError(code, msg)));
                 }
diff --git a/src/main/resources/openapi.json b/src/main/resources/openapi.json
index 51696d9..2ff94db 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": [
                     {
@@ -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..804b0c2 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": [
         {
@@ -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/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 a491903..9858c73 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ProducerTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ProducerTest.java
@@ -599,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/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()))