Add a batch version of publishToKafka (#88)

* Add a batch version of publishToKafka

It is very expensive to create a `KafkaProducer`.

For tests that send a lot of messages it is very slow
to use the original `publishToKafka` method that created
a new `KafkaProducer` for each message to send.

Now there is a batch version of `publishToKafka` that
accepts a sequence of key/value tuples and sends them
all with the same `KafkaProducer` instance.

* Replace `filter` with `find`

Author: blissd

embedded-kafka/src/main/scala/net/manub/embeddedkafka/EmbeddedKafka.scala

@@ -247,8 +247,40 @@ sealed trait EmbeddedKafkaSupport {
       new KafkaProducer(baseProducerConfig.asJava, keySerializer, serializer),
       new ProducerRecord(topic, key, message))
 
+
+  /**
+    * Publishes synchronously a batch of message to the running Kafka broker.
+    *
+    * @param topic      the topic to which publish the message (it will be auto-created)
+    * @param messages    the keys and messages of type [[(K, T)]] to publish
+    * @param config     an implicit [[EmbeddedKafkaConfig]]
+    * @param keySerializer an implicit [[Serializer]] for the type [[K]]
+    * @param serializer an implicit [[Serializer]] for the type [[T]]
+    * @throws KafkaUnavailableException if unable to connect to Kafka
+    */
+  @throws(classOf[KafkaUnavailableException])
+  def publishToKafka[K, T](topic: String, messages: Seq[(K, T)])(
+    implicit config: EmbeddedKafkaConfig,
+    keySerializer: Serializer[K],
+    serializer: Serializer[T]): Unit = {
+
+    val producer = new KafkaProducer(baseProducerConfig.asJava, keySerializer, serializer)
+
+    val tupleToRecord = (new ProducerRecord(topic, _: K, _: T)).tupled
+
+    val futureSend = tupleToRecord andThen producer.send
+
+    val futures = messages.map(futureSend)
+
+    // Assure all messages sent before returning, and fail on first send error
+    val records = futures.map(f => Try(f.get(10, SECONDS)))
+    records.find(_.isFailure).foreach(record => throw new KafkaUnavailableException(record.failed.get))
+
+    producer.close()
+  }
+
   private def publishToKafka[K, T](kafkaProducer: KafkaProducer[K, T],
-                                   record: ProducerRecord[K, T]) = {
+                                   record: ProducerRecord[K, T]): Unit = {
     val sendFuture = kafkaProducer.send(record)
     val sendResult = Try {
       sendFuture.get(10, SECONDS)

embedded-kafka/src/test/scala/net/manub/embeddedkafka/EmbeddedKafkaMethodsSpec.scala

@@ -79,6 +79,38 @@ class EmbeddedKafkaMethodsSpec
 
       consumer.close()
     }
+
+    "publish synchronously a batch of String messages with String keys to Kafka" in {
+      implicit val serializer = new StringSerializer()
+      implicit val deserializer = new StringDeserializer()
+      val key1 = "key1"
+      val message1 = "hello world!"
+      val key2 = "key2"
+      val message2 = "goodbye world!"
+      val topic = "publish_test_topic_batch_string_key"
+
+      val messages = List((key1, message1), (key2, message2))
+
+      publishToKafka(topic, messages)
+
+      val consumer = kafkaConsumer
+      consumer.subscribe(List(topic).asJava)
+
+      val records = consumer.poll(consumerPollTimeout).iterator()
+
+      records.hasNext shouldBe true
+
+      val record1 = records.next()
+      record1.key() shouldBe key1
+      record1.value() shouldBe message1
+
+      records.hasNext shouldBe true
+      val record2 = records.next()
+      record2.key() shouldBe key2
+      record2.value() shouldBe message2
+
+      consumer.close()
+    }
   }
 
   "the createCustomTopic method" should {