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

Issue:
Support for sending multiple records in one HTTP POST
The current HTTP POST has support for sending a single record while we should have support for sending multiple records in one request.
It's also about providing an HTTP response with metadata not for only one record but for the multiple records sent.

Pull request description:
Added sending more records via one HTTP POST request
This PR fixes #111

Patch:
diff --git a/doc/HTTP_Producer_API.md b/doc/HTTP_Producer_API.md
index 4944a19..98e2f0e 100644
--- a/doc/HTTP_Producer_API.md
+++ b/doc/HTTP_Producer_API.md
@@ -11,25 +11,34 @@ Producer API allows you to send message to a topic, optionally specifying key or
 
 Request body should be in JSON format.
 
-Sample request specifying a key:
+Sample request for sending three records:
 
-```
-POST /topics/kafka-bridge HTTP/1.1
-Host: localhost:8080
-Content-Type: application/application/json
-{"key":"my-key","value":"Hi this is kafka-bridge"}
-```
-
-Sample request specifying a partition:
+* specifying a key
+* specifying a partition
+* just with message value
 
 ```
 POST /topics/kafka-bridge HTTP/1.1
 Host: localhost:8080
 Content-Type: application/application/json
-{"partition":1,"value":"Hi this is kafka-bridge"}
+{
+    "records": [
+        {
+            "key": "my-key",
+            "value": "Hi this is kafka-bridge (with key)"
+        },
+        {
+            "value": "Hi this is kafka-bridge (with partition)",
+            "partition": 1
+        },
+        {
+            "value": "Hi this is kafka-bridge"
+        }
+    ]
+}
 ```
 
-Response is also a JSON object containing delivery report and metadata of the produced message.
+Response is also a JSON object containing metadata of the produced messages.
 
 Sample response
 
@@ -37,7 +46,22 @@ Sample response
 HTTP/1.1 200 OK
 Content-Type: application/json
 
-{"status":"Accepted","topic":"kafka-bridge","partition": 1,"offset": 1}
+{
+    "offsets": [
+        {
+            "partition": 2,
+            "offset": 0
+        },
+        {
+            "partition": 1,
+            "offset": 0
+        },
+        {
+            "partition": 2,
+            "offset": 1
+        }
+    ]
+}
 ```
 
-In case of errors while producing message, the `status` will be sent as `Rejected`.
+In case of errors the corresponding "offset" JSON contains "error_code" and "error" fields.
\ No newline at end of file
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpDefaultMessageConverter.java b/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpDefaultMessageConverter.java
index df6fee3..6fa2cfe 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpDefaultMessageConverter.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpDefaultMessageConverter.java
@@ -30,6 +30,7 @@ import org.apache.qpid.proton.amqp.messaging.Data;
 import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
 import org.apache.qpid.proton.amqp.messaging.Section;
 import org.apache.qpid.proton.message.Message;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
 
 import java.util.Collection;
 import java.util.HashMap;
@@ -134,7 +135,11 @@ public class AmqpDefaultMessageConverter implements MessageConverter<String, byt
 
 	@Override
 	public Collection<Message> toMessages(KafkaConsumerRecords<String, byte[]> records) {
-		return null;
+		throw new NotImplementedException();
 	}
 
+	@Override
+	public List<KafkaProducerRecord<String, byte[]>> toKafkaRecords(String kafkaTopic, Collection<Message> messages) {
+		throw new NotImplementedException();
+	}
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpJsonMessageConverter.java b/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpJsonMessageConverter.java
index 5128974..55456cb 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpJsonMessageConverter.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpJsonMessageConverter.java
@@ -28,6 +28,7 @@ import org.apache.qpid.proton.amqp.Binary;
 import org.apache.qpid.proton.amqp.Symbol;
 import org.apache.qpid.proton.amqp.messaging.*;
 import org.apache.qpid.proton.message.Message;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
 
 import java.util.*;
 import java.util.Map.Entry;
@@ -286,8 +287,11 @@ public class AmqpJsonMessageConverter implements MessageConverter<String, byte[]
 
 	@Override
 	public Collection<Message> toMessages(KafkaConsumerRecords<String, byte[]> records) {
-		return null;
+		throw new NotImplementedException();
 	}
 
-
+	@Override
+	public List<KafkaProducerRecord<String, byte[]>> toKafkaRecords(String kafkaTopic, Collection<Message> messages) {
+		throw new NotImplementedException();
+	}
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpRawMessageConverter.java b/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpRawMessageConverter.java
index bfffa66..a4379e4 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpRawMessageConverter.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/converter/AmqpRawMessageConverter.java
@@ -25,10 +25,12 @@ import org.apache.qpid.proton.Proton;
 import org.apache.qpid.proton.amqp.Symbol;
 import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
 import org.apache.qpid.proton.message.Message;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
 
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -100,7 +102,11 @@ public class AmqpRawMessageConverter implements MessageConverter<String, byte[],
 
 	@Override
 	public Collection<Message> toMessages(KafkaConsumerRecords<String, byte[]> records) {
-		return null;
+		throw new NotImplementedException();
 	}
 
+	@Override
+	public List<KafkaProducerRecord<String, byte[]>> toKafkaRecords(String kafkaTopic, Collection<Message> messages) {
+		throw new NotImplementedException();
+	}
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/converter/MessageConverter.java b/src/main/java/io/strimzi/kafka/bridge/converter/MessageConverter.java
index 7ee84c0..e0dcf61 100644
--- a/src/main/java/io/strimzi/kafka/bridge/converter/MessageConverter.java
+++ b/src/main/java/io/strimzi/kafka/bridge/converter/MessageConverter.java
@@ -16,38 +16,49 @@
 
 package io.strimzi.kafka.bridge.converter;
 
-
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecords;
 import io.vertx.kafka.client.producer.KafkaProducerRecord;
 
+import java.util.List;
+
 /**
- * Interface for a message converter between Kafka record and AMQP message
+ * Interface for a message converter between Kafka record and bridge message
  */
 public interface MessageConverter<K, V, M, C> {
 
 	/**
 	 * Converts a message to a Kafka record
 	 *
-	 * @param kafkaTopic	Kafka topic for sending message
-	 * @param message	message to convert
-	 * @return	Kafka record
+	 * @param kafkaTopic Kafka topic for sending message
+	 * @param message message to convert
+	 * @return Kafka record
 	 */
 	KafkaProducerRecord<K, V> toKafkaRecord(String kafkaTopic, M message);
+
+	/**
+	 * Convert a collection of messages to Kafka records
+	 *
+	 * @param kafkaTopic Kafka topic for sending message
+	 * @param messages collection of messages to convert
+	 * @return Kafka records
+	 */
+	List<KafkaProducerRecord<K, V>> toKafkaRecords(String kafkaTopic, C messages);
 	
 	/**
 	 * Converts a Kafka record to a message
 	 *
-	 * @param address	address for sending message
-	 * @param record	Kafka record to convert
-	 * @return	message
+	 * @param address address for sending message
+	 * @param record Kafka record to convert
+	 * @return message
 	 */
 	M toMessage(String address, KafkaConsumerRecord<K, V> record);
 
 	/**
-	 * Converts multiple KafkaConsumerRecord from KafkaConsumerRecords to a message
-	 * @param records KafkaRecords to convert
-	 * @return a list of messages
+	 * Converts Kafka records to a collection of messages
+	 *
+	 * @param records Kafka records to convert
+	 * @return a collection of messages
 	 */
 	C toMessages(KafkaConsumerRecords<K, V> records);
 }
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 4bca7cc..ba03c08 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
@@ -20,14 +20,22 @@ import io.strimzi.kafka.bridge.Endpoint;
 import io.strimzi.kafka.bridge.SourceBridgeEndpoint;
 import io.strimzi.kafka.bridge.converter.MessageConverter;
 import io.strimzi.kafka.bridge.http.converter.HttpJsonMessageConverter;
+import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
+import io.strimzi.kafka.bridge.http.model.HttpBridgeResult;
+import io.vertx.core.CompositeFuture;
+import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.http.HttpServerRequest;
 import io.vertx.core.http.HttpServerResponse;
+import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.kafka.client.producer.KafkaProducerRecord;
 import io.vertx.kafka.client.producer.RecordMetadata;
 
+import java.util.ArrayList;
+import java.util.List;
+
 public class HttpSourceBridgeEndpoint extends SourceBridgeEndpoint {
     private MessageConverter messageConverter;
 
@@ -41,31 +49,41 @@ public class HttpSourceBridgeEndpoint extends SourceBridgeEndpoint {
 
         messageConverter = new HttpJsonMessageConverter();
 
-        //split path to extract params
+        // split path to extract params
         String[] params = httpServerRequest.path().split("/");
 
-        //path is like this : /topics/{topic_name}, topic will be at the last position of param[]
+        // path is like this : /topics/{topic_name}, topic will be at the last position of params
         String topic = params[params.length - 1];
 
         httpServerRequest.bodyHandler(buffer -> {
-            KafkaProducerRecord<String , byte[]> kafkaProducerRecord = messageConverter.toKafkaRecord(topic, buffer);
-
-            this.send(kafkaProducerRecord, writeResult -> {
-                if (writeResult.failed()) {
-
-                    log.error("Error on delivery to Kafka {}", writeResult.cause());
-                    this.sendRejectedDeliveryResponse(httpServerRequest.response());
-
-                } else {
-                    RecordMetadata metadata = writeResult.result();
-                    log.debug("Delivered to Kafka on topic {} at partition {} [{}]", metadata.getTopic(), metadata.getPartition(), metadata.getOffset());
-                    this.sendAcceptedDeliveryResponse(metadata, httpServerRequest.response());
-
+            List<KafkaProducerRecord<String , byte[]>> records = messageConverter.toKafkaRecords(topic, buffer);
+            List<HttpBridgeResult<?>> results = new ArrayList<>(records.size());
+
+            // start sending records asynchronously
+            List<Future> sendHandlers = new ArrayList<>(records.size());
+            for (KafkaProducerRecord<String, byte[]> record : records) {
+                Future<RecordMetadata> fut = Future.future();
+                sendHandlers.add(fut);
+                this.send(record, fut.completer());
+            }
+
+            // wait for ALL futures completed
+            CompositeFuture.join(sendHandlers).setHandler(done -> {
+                for (int i = 0; i < sendHandlers.size(); i++) {
+                    // check if, for each future, the sending operation is completed successfully or failed
+                    if (done.result().succeeded(i)) {
+                        RecordMetadata metadata = done.result().resultAt(i);
+                        log.debug("Delivered record {} to Kafka on topic {} at partition {} [{}]", records.get(i), metadata.getTopic(), metadata.getPartition(), metadata.getOffset());
+                        results.add(new HttpBridgeResult<>(metadata));
+                    } else {
+                        log.error("Failed to deliver record " + records.get(i) + " due to {}", done.result().cause(i));
+                        // TODO: error codes definition
+                        results.add(new HttpBridgeResult<>(new HttpBridgeError(0, done.result().cause(i).getMessage())));
+                    }
                 }
+                sendMetadataResponse(results, httpServerRequest.response());
             });
-
         });
-
     }
 
     @Override
@@ -73,22 +91,24 @@ public class HttpSourceBridgeEndpoint extends SourceBridgeEndpoint {
 
     }
 
-    private void sendAcceptedDeliveryResponse(RecordMetadata metadata, HttpServerResponse response){
-
-        JsonObject jsonResponse = new JsonObject();
-        jsonResponse.put("status", "Accepted");
-        jsonResponse.put("topic", metadata.getTopic());
-        jsonResponse.put("partition", metadata.getPartition());
-        jsonResponse.put("offset", metadata.getOffset());
-
-        response.putHeader("Content-length", String.valueOf(jsonResponse.toBuffer().length()));
-        response.write(jsonResponse.toBuffer());
-        response.end();
-    }
-
-    private void sendRejectedDeliveryResponse(HttpServerResponse response){
+    private void sendMetadataResponse(List<HttpBridgeResult<?>> results, HttpServerResponse response) {
         JsonObject jsonResponse = new JsonObject();
-        jsonResponse.put("status", "rejected");
+        JsonArray offsets = new JsonArray();
+
+        for (HttpBridgeResult<?> result : results) {
+            JsonObject offset = new JsonObject();
+            if (result.getResult() instanceof RecordMetadata) {
+                RecordMetadata metadata = (RecordMetadata) result.getResult();
+                offset.put("partition", metadata.getPartition());
+                offset.put("offset", metadata.getOffset());
+            } else if (result.getResult() instanceof HttpBridgeError) {
+                HttpBridgeError error = (HttpBridgeError) result.getResult();
+                offset.put("error_code", error.getCode());
+                offset.put("error", error.getMessage());
+            }
+            offsets.add(offset);
+        }
+        jsonResponse.put("offsets", offsets);
 
         response.putHeader("Content-length", String.valueOf(jsonResponse.toBuffer().length()));
         response.write(jsonResponse.toBuffer());
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpJsonMessageConverter.java b/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpJsonMessageConverter.java
index c5ffdfb..3ee12ef 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpJsonMessageConverter.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpJsonMessageConverter.java
@@ -23,8 +23,12 @@ import io.vertx.core.json.JsonObject;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecord;
 import io.vertx.kafka.client.consumer.KafkaConsumerRecords;
 import io.vertx.kafka.client.producer.KafkaProducerRecord;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
 
-public class HttpJsonMessageConverter implements MessageConverter<String,byte[],Buffer, Buffer> {
+import java.util.ArrayList;
+import java.util.List;
+
+public class HttpJsonMessageConverter implements MessageConverter<String, byte[], Buffer, Buffer> {
 
     @Override
     public KafkaProducerRecord<String, byte[]> toKafkaRecord(String kafkaTopic, Buffer message) {
@@ -51,9 +55,25 @@ public class HttpJsonMessageConverter implements MessageConverter<String,byte[],
         return record;
     }
 
+    @Override
+    public List<KafkaProducerRecord<String, byte[]>> toKafkaRecords(String kafkaTopic, Buffer messages) {
+
+        List<KafkaProducerRecord<String, byte[]>> records = new ArrayList<>();
+
+        JsonObject json = messages.toJsonObject();
+        JsonArray jsonArray = json.getJsonArray("records");
+
+        for (Object obj : jsonArray) {
+            JsonObject jsonObj = (JsonObject) obj;
+            records.add(toKafkaRecord(kafkaTopic, jsonObj.toBuffer()));
+        }
+
+        return records;
+    }
+
     @Override
     public Buffer toMessage(String address, KafkaConsumerRecord<String, byte[]> record) {
-        return null;
+        throw new NotImplementedException();
     }
 
     @Override
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeError.java b/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeError.java
new file mode 100644
index 0000000..4dff064
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeError.java
@@ -0,0 +1,51 @@
+/*
+ * Copyright 2019 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.strimzi.kafka.bridge.http.model;
+
+/**
+ * Represents an error related to HTTP bridging
+ */
+public class HttpBridgeError {
+
+    private final int code;
+    private final String message;
+
+    /**
+     * Constructor
+     *
+     * @param code code classifying the error itself
+     * @param message message providing more information about the error
+     */
+    public HttpBridgeError(int code, String message) {
+        this.code = code;
+        this.message = message;
+    }
+
+    /**
+     * @return code classifying the error itself
+     */
+    public int getCode() {
+        return code;
+    }
+
+    /**
+     * @return message providing more information about the error
+     */
+    public String getMessage() {
+        return message;
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeResult.java b/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeResult.java
new file mode 100644
index 0000000..d679528
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/http/model/HttpBridgeResult.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2019 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.strimzi.kafka.bridge.http.model;
+
+/**
+ * This class represents a result of an HTTP bridging operation
+ *
+ * @param <T> the class bringing the actual result as {@link HttpBridgeError} or {@link io.vertx.kafka.client.producer.RecordMetadata}
+ */
+public class HttpBridgeResult<T> {
+
+    final T result;
+
+    /**
+     * Constructor
+     *
+     * @param result actual result
+     */
+    public HttpBridgeResult(T result) {
+        this.result = result;
+    }
+
+    /**
+     * @return the actual result
+     */
+    public T getResult() {
+        return result;
+    }
+}


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
index 1940c74..ac4985c 100644
--- a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
@@ -466,6 +466,11 @@ public class AmqpSinkBridgeEndpointMockTest {
 			return null;
 		}
 
+		@Override
+		public List<KafkaProducerRecord<K, V>>toKafkaRecords(String kafkaTopic, C messages) {
+			return null;
+		}
+
 		@Override
 		public M toMessage(String address, KafkaConsumerRecord<K, V> record) {
 			return null;
@@ -503,6 +508,11 @@ public class AmqpSinkBridgeEndpointMockTest {
 			return null;
 		}
 
+		@Override
+		public List<KafkaProducerRecord<K, V>> toKafkaRecords(String kafkaTopic, C messages) {
+			return null;
+		}
+
 		@Override
 		public M toMessage(String address, KafkaConsumerRecord<K, V> record) {
 			return null;
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
index f5f13f0..8d3fa3e 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
@@ -54,9 +54,10 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
     private static final String BRIDGE_HOST = "0.0.0.0";
     private static final int BRIDGE_PORT = 8080;
 
-    // for periodic test
+    // for periodic/multiple messages test
     private static final int PERIODIC_MAX_MESSAGE = 10;
     private static final int PERIODIC_DELAY = 200;
+    private static final int MULTIPLE_MAX_MESSAGE = 10;
     private int count;
 
     private Vertx vertx;
@@ -83,38 +84,37 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
     @Test
     public void sendSimpleMessage(TestContext context) {
-        String kafkaTopic = "sendSimpleMessage";
-        kafkaCluster.createTopic(kafkaTopic, 1, 1);
+        String topic = "sendSimpleMessage";
+        kafkaCluster.createTopic(topic, 1, 1);
 
         Async async = context.async();
 
-        String value = "Hi, This is kafka bridge";
+        String value = "message-value";
 
+        JsonArray records = new JsonArray();
         JsonObject json = new JsonObject();
         json.put("value", value);
+        records.add(json);
+
+        JsonObject root = new JsonObject();
+        root.put("records", records);
 
         WebClient client = WebClient.create(vertx);
 
-        client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + kafkaTopic)
-                .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+        client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + topic)
+                .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
                 .as(BodyCodec.jsonObject())
-                .sendJsonObject(json, ar -> {
+                .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
 
                     HttpResponse<JsonObject> response = ar.result();
                     JsonObject bridgeResponse = response.body();
-                    String deliveryStatus = bridgeResponse.getString("status");
-                    String topic = bridgeResponse.getString("topic");
-                    String key = bridgeResponse.getString("key");
-                    long offset = bridgeResponse.getLong("offset");
-                    //check delivery status
-                    context.assertEquals("Accepted", deliveryStatus);
-                    //check topic
-                    context.assertEquals(kafkaTopic, topic);
-                    //check offset. should be 0 as single message is published
-                    context.assertEquals(0L, offset);
-                    //check key. should be null
-                    context.assertNull(key);
+
+                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                    context.assertEquals(1, offsets.size());
+                    JsonObject metadata = offsets.getJsonObject(0);
+                    context.assertEquals(0, metadata.getInteger("partition"));
+                    context.assertEquals(0L, metadata.getLong("offset"));
                 });
 
         Properties config = kafkaCluster.useTo().getConsumerProperties("groupId", null, OffsetResetStrategy.EARLIEST);
@@ -123,9 +123,10 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config);
         consumer.handler(record -> {
-            context.assertEquals(record.value(), value);
-            context.assertEquals(record.topic(), kafkaTopic);
-            context.assertEquals(record.offset(), 0L);
+            context.assertEquals(value, record.value());
+            context.assertEquals(topic, record.topic());
+            context.assertEquals(0, record.partition());
+            context.assertEquals(0L, record.offset());
             context.assertNull(record.key());
             log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                     record.topic(), record.partition(), record.offset(), record.key(), record.value());
@@ -133,7 +134,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
             async.complete();
         });
 
-        consumer.subscribe(kafkaTopic, done -> {
+        consumer.subscribe(topic, done -> {
             if (!done.succeeded()) {
                 context.fail(done.cause());
             }
@@ -142,42 +143,40 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
     @Test
     public void sendSimpleMessageToPartition(TestContext context) {
-        String kafkaTopic = "sendSimpleMessageToPartition";
-
-        kafkaCluster.createTopic(kafkaTopic, 2, 1);
+        String topic = "sendSimpleMessageToPartition";
+        kafkaCluster.createTopic(topic, 2, 1);
 
         Async async = context.async();
 
-        String value = "Hi, This is kafka bridge";
+        String value = "message-value";
 
-        int kafkaPartition = 1;
+        int partition = 1;
 
+        JsonArray records = new JsonArray();
         JsonObject json = new JsonObject();
         json.put("value", value);
-        json.put("partition", kafkaPartition);
+        json.put("partition", partition);
+        records.add(json);
+
+        JsonObject root = new JsonObject();
+        root.put("records", records);
 
         WebClient client = WebClient.create(vertx);
 
-        client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + kafkaTopic)
-                .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+        client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + topic)
+                .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
                 .as(BodyCodec.jsonObject())
-                .sendJsonObject(json, ar -> {
+                .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
 
                     HttpResponse<JsonObject> response = ar.result();
                     JsonObject bridgeResponse = response.body();
-                    String deliveryStatus = bridgeResponse.getString("status");
-                    String topic = bridgeResponse.getString("topic");
-                    int partition = bridgeResponse.getInteger("partition");
-                    long offset = bridgeResponse.getLong("offset");
-                    //check delivery status
-                    context.assertEquals("Accepted", deliveryStatus);
-                    //check topic
-                    context.assertEquals(kafkaTopic, topic);
-                    //check partition
-                    context.assertEquals(kafkaPartition, partition);
-                    //check offset. should be 0 as single message is published
-                    context.assertEquals(0L, offset);
+
+                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                    context.assertEquals(1, offsets.size());
+                    JsonObject metadata = offsets.getJsonObject(0);
+                    context.assertEquals(partition, metadata.getInteger("partition"));
+                    context.assertEquals(0L, metadata.getLong("offset"));
                 });
 
         Properties config = kafkaCluster.useTo().getConsumerProperties("groupId", null, OffsetResetStrategy.EARLIEST);
@@ -186,11 +185,10 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config);
         consumer.handler(record -> {
-            context.assertEquals(record.value(), value);
-            //should be from same partition
-            context.assertEquals(record.partition(), kafkaPartition);
-            context.assertEquals(record.topic(), kafkaTopic);
-            context.assertEquals(record.offset(), 0L);
+            context.assertEquals(value, record.value());
+            context.assertEquals(topic, record.topic());
+            context.assertEquals(partition, record.partition());
+            context.assertEquals(0L, record.offset());
             context.assertNull(record.key());
             log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                     record.topic(), record.partition(), record.offset(), record.key(), record.value());
@@ -198,7 +196,7 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
             async.complete();
         });
 
-        consumer.subscribe(kafkaTopic, done -> {
+        consumer.subscribe(topic, done -> {
             if (!done.succeeded()) {
                 context.fail(done.cause());
             }
@@ -207,39 +205,39 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
     @Test
     public void sendSimpleMessageWithKey(TestContext context) {
-        String kafkaTopic = "sendSimpleMessageWithKey";
-
-        kafkaCluster.createTopic(kafkaTopic, 2, 1);
+        String topic = "sendSimpleMessageWithKey";
+        kafkaCluster.createTopic(topic, 2, 1);
 
         Async async = context.async();
 
-        String value = "Hi, This is kafka bridge";
-
-        String kafkaKey = "my_key";
+        String value = "message-value";
+        String key = "my-key";
 
+        JsonArray records = new JsonArray();
         JsonObject json = new JsonObject();
         json.put("value", value);
-        json.put("key", kafkaKey);
+        json.put("key", key);
+        records.add(json);
+
+        JsonObject root = new JsonObject();
+        root.put("records", records);
 
         WebClient client = WebClient.create(vertx);
 
-        client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + kafkaTopic)
-                .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
+        client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + topic)
+                .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
                 .as(BodyCodec.jsonObject())
-                .sendJsonObject(json, ar -> {
+                .sendJsonObject(root, ar -> {
                     context.assertTrue(ar.succeeded());
 
                     HttpResponse<JsonObject> response = ar.result();
                     JsonObject bridgeResponse = response.body();
-                    String deliveryStatus = bridgeResponse.getString("status");
-                    String topic = bridgeResponse.getString("topic");
-                    long offset = bridgeResponse.getLong("offset");
-                    //check delivery status
-                    context.assertEquals("Accepted", deliveryStatus);
-                    //check topic
-                    context.assertEquals(kafkaTopic, topic);
-                    //check offset. should be 0 as single message is published
-                    context.assertEquals(0L, offset);
+
+                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                    context.assertEquals(1, offsets.size());
+                    JsonObject metadata = offsets.getJsonObject(0);
+                    context.assertNotNull(metadata.getInteger("partition"));
+                    context.assertEquals(0L, metadata.getLong("offset"));
                 });
 
         Properties config = kafkaCluster.useTo().getConsumerProperties("groupId", null, OffsetResetStrategy.EARLIEST);
@@ -248,17 +246,18 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
         KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config);
         consumer.handler(record -> {
-            context.assertEquals(record.value(), value);
-            context.assertEquals(record.key(), kafkaKey);
-            context.assertEquals(record.topic(), kafkaTopic);
-            context.assertEquals(record.offset(), 0L);
+            context.assertEquals(value, record.value());
+            context.assertEquals(topic, record.topic());
+            context.assertNotNull(record.partition());
+            context.assertEquals(0L, record.offset());
+            context.assertEquals(record.key(), key);
             log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                     record.topic(), record.partition(), record.offset(), record.key(), record.value());
             consumer.close();
             async.complete();
         });
 
-        consumer.subscribe(kafkaTopic, done -> {
+        consumer.subscribe(topic, done -> {
             if (!done.succeeded()) {
                 context.fail(done.cause());
             }
@@ -286,13 +285,18 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
             if (this.count < HttpBridgeTest.PERIODIC_MAX_MESSAGE){
 
+                JsonArray records = new JsonArray();
                 JsonObject json = new JsonObject();
                 json.put("value", "Periodic message [" + this.count + "]");
                 json.put("key", "key-" + this.count);
+                records.add(json);
+
+                JsonObject root = new JsonObject();
+                root.put("records", records);
 
                 client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + topic)
-                        .putHeader("Content-length", String.valueOf(json.toBuffer().length()))
-                        .sendJsonObject(json, ar -> { });
+                        .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
+                        .sendJsonObject(root, ar -> { });
 
                 this.count++;
             } else {
@@ -312,8 +316,11 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
                 KafkaConsumerRecord<String, String> record = records.recordAt(i);
                 log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
                         record.topic(), record.partition(), record.offset(), record.key(), record.value());
-                context.assertEquals("key-" + i, record.key());
-                context.assertEquals("Periodic message [" + i + "]", record.value());
+                context.assertEquals(record.value(), "Periodic message [" + i + "]");
+                context.assertEquals(topic, record.topic());
+                context.assertNotNull(record.partition());
+                context.assertNotNull(record.offset());
+                context.assertEquals(record.key(), "key-" + i);
             }
 
             consumer.close();
@@ -323,6 +330,74 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
         consumer.handler(record -> {});
     }
 
+    @Test
+    public void sendMultipleMessages(TestContext context) {
+        String topic = "sendMultipleMessages";
+        kafkaCluster.createTopic(topic, 1, 1);
+
+        Async async = context.async();
+
+        String value = "message-value";
+
+        int numMessages = MULTIPLE_MAX_MESSAGE;
+
+        JsonArray records = new JsonArray();
+        for (int i = 0; i < numMessages; i++) {
+            JsonObject json = new JsonObject();
+            json.put("value", value + "-" + i);
+            records.add(json);
+        }
+        JsonObject root = new JsonObject();
+        root.put("records", records);
+
+        WebClient client = WebClient.create(vertx);
+
+        client.post(BRIDGE_PORT, BRIDGE_HOST, "/topics/" + topic)
+                .putHeader("Content-length", String.valueOf(root.toBuffer().length()))
+                .as(BodyCodec.jsonObject())
+                .sendJsonObject(root, ar -> {
+                    context.assertTrue(ar.succeeded());
+
+                    HttpResponse<JsonObject> response = ar.result();
+                    JsonObject bridgeResponse = response.body();
+
+                    JsonArray offsets = bridgeResponse.getJsonArray("offsets");
+                    context.assertEquals(numMessages, offsets.size());
+                    for (int i = 0; i < numMessages; i++) {
+                        JsonObject metadata = offsets.getJsonObject(i);
+                        context.assertEquals(0, metadata.getInteger("partition"));
+                        context.assertNotNull(metadata.getLong("offset"));
+                    }
+                });
+
+        Properties config = kafkaCluster.useTo().getConsumerProperties("groupId", null, OffsetResetStrategy.EARLIEST);
+        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+
+        KafkaConsumer<String, String> consumer = KafkaConsumer.create(this.vertx, config);
+        this.count = 0;
+        consumer.handler(record -> {
+            context.assertEquals(value + "-" + this.count++, record.value());
+            context.assertEquals(topic, record.topic());
+            context.assertNotNull(record.partition());
+            context.assertNotNull(record.offset());
+            context.assertNull(record.key());
+            log.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
+                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
+
+            if (this.count == numMessages) {
+                consumer.close();
+                async.complete();
+            }
+        });
+
+        consumer.subscribe(topic, done -> {
+            if (!done.succeeded()) {
+                context.fail(done.cause());
+            }
+        });
+    }
+
     @Test
     public void receiveSimpleMessage(TestContext context) {
         String topic = "receiveSimpleMessage";