Allow consumeFirstStringMessageFrom to be called multiple times

In v0.5.0 you were able to call consumeFirstStringMessageFrom multiple
times, with each invocation causing the next message in a topic to be
consumed. In v0.10.0 this was no longer the case: the first invocation
works fine, but subsequent invocations would result in exceptions as the
first invocation would cause *all* messages to be committed even though
only the first one is returned, leaving no messages remaining to be
consumed by subsequent calls. This regression was likely introduced by
commit a906743.

This commit fixes this by disabling auto-commit by default and simply
committing only the message that consumeFirstStringMessageFrom returns.

Author: Alex Adriaanse

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

@@ -7,9 +7,9 @@ import java.util.concurrent.Executors
 import kafka.admin.AdminUtils
 import kafka.server.{KafkaConfig, KafkaServer}
 import kafka.utils.ZkUtils
-import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.clients.consumer.{KafkaConsumer, OffsetAndMetadata}
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
-import org.apache.kafka.common.KafkaException
+import org.apache.kafka.common.{KafkaException, TopicPartition}
 import org.apache.kafka.common.serialization.{Deserializer, Serializer, StringDeserializer, StringSerializer}
 import org.apache.zookeeper.server.{ServerCnxnFactory, ZooKeeperServer}
 import org.scalatest.Suite
@@ -217,6 +217,8 @@ sealed trait EmbeddedKafkaSupport {
   /**
     * Consumes the first message available in a given topic, deserializing it as a String.
     *
+    * Only the messsage that is returned is committed if config.autoCommit is false. If config.autoCommit is true then all messages that were polled will be committed.
+    *
     * @param topic        the topic to consume a message from
     * @param config       an implicit [[EmbeddedKafkaConfig]]
     * @param deserializer an implicit [[org.apache.kafka.common.serialization.Deserializer]] for the type [[T]]
@@ -236,6 +238,7 @@ sealed trait EmbeddedKafkaSupport {
     props.put("group.id", s"embedded-kafka-spec")
     props.put("bootstrap.servers", s"localhost:${config.kafkaPort}")
     props.put("auto.offset.reset", "earliest")
+    props.put("enable.auto.commit", s"${config.autoCommit}")
 
     val consumer =
       new KafkaConsumer[String, T](props, new StringDeserializer, deserializer)
@@ -248,7 +251,14 @@ sealed trait EmbeddedKafkaSupport {
         throw new TimeoutException(
           "Unable to retrieve a message from Kafka in 5000ms")
       }
-      records.iterator().next().value()
+
+      val record = records.iterator().next()
+
+      val tp = new TopicPartition(record.topic(), record.partition())
+      val om = new OffsetAndMetadata(record.offset() + 1)
+      consumer.commitSync(Map(tp -> om))
+
+      record.value()
     }
 
     consumer.close()

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

@@ -2,7 +2,8 @@ package net.manub.embeddedkafka
 
 case class EmbeddedKafkaConfig(kafkaPort: Int = 6001,
                                zooKeeperPort: Int = 6000,
-                               customBrokerProperties: Map[String, String] = Map.empty)
+                               customBrokerProperties: Map[String, String] = Map.empty,
+                               autoCommit: Boolean = false)
 
 object EmbeddedKafkaConfig {
   implicit val defaultConfig = EmbeddedKafkaConfig()

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

@@ -28,12 +28,12 @@ class EmbeddedKafkaMethodsSpec extends EmbeddedKafkaSpecSupport with EmbeddedKaf
     "publish synchronously a String message to Kafka" in {
       implicit val serializer = new StringSerializer()
       val message = "hello world!"
-      val topic = "test_topic"
+      val topic = "publish_test_topic"
 
       publishToKafka(topic, message)
 
       val consumer = new KafkaConsumer[String, String](consumerProps, new StringDeserializer, new StringDeserializer)
-      consumer.subscribe(List("test_topic"))
+      consumer.subscribe(List(topic))
 
       val records = consumer.poll(ConsumerPollTimeout)
 
@@ -50,12 +50,12 @@ class EmbeddedKafkaMethodsSpec extends EmbeddedKafkaSpecSupport with EmbeddedKaf
       implicit val serializer = new StringSerializer()
       val key = "key"
       val message = "hello world!"
-      val topic = "test_topic"
+      val topic = "publish_test_topic"
 
       publishToKafka(topic, key, message)
 
       val consumer = new KafkaConsumer[String, String](consumerProps, new StringDeserializer, new StringDeserializer)
-      consumer.subscribe(List("test_topic"))
+      consumer.subscribe(List(topic))
 
       val records = consumer.poll(ConsumerPollTimeout)
 
@@ -111,7 +111,7 @@ class EmbeddedKafkaMethodsSpec extends EmbeddedKafkaSpecSupport with EmbeddedKaf
   "the consumeFirstStringMessageFrom method" should {
     "return a message published to a topic" in {
       val message = "hello world!"
-      val topic = "test_topic"
+      val topic = "consume_test_topic"
 
       val producer = new KafkaProducer[String, String](Map(
         ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> s"localhost:6001",
@@ -126,9 +126,34 @@ class EmbeddedKafkaMethodsSpec extends EmbeddedKafkaSpecSupport with EmbeddedKaf
       producer.close()
     }
 
+    "consume only a single message when multiple messages have been published to a topic" in {
+      val messages = Set("message 1", "message 2", "message 3")
+      val topic = "consume_test_topic"
+
+      val producer = new KafkaProducer[String, String](Map(
+        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> s"localhost:6001",
+        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG -> classOf[StringSerializer].getName,
+        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG -> classOf[StringSerializer].getName
+      ))
+
+      messages.foreach { message =>
+        producer.send(new ProducerRecord[String, String](topic, message))
+      }
+
+      producer.flush()
+
+      val consumedMessages = for (i <- 1 to messages.size) yield {
+        consumeFirstStringMessageFrom(topic)
+      }
+
+      consumedMessages.toSet shouldEqual messages
+
+      producer.close()
+    }
+
     "return a message published to a topic with implicit decoder" in {
       val message = "hello world!"
-      val topic = "test_topic"
+      val topic = "consume_test_topic"
 
       val producer = new KafkaProducer[String, String](Map(
         ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> s"localhost:6001",
@@ -149,7 +174,7 @@ class EmbeddedKafkaMethodsSpec extends EmbeddedKafkaSpecSupport with EmbeddedKaf
       import avro._
 
       val message = TestAvroClass("name")
-      val topic = "test_topic"
+      val topic = "consume_test_topic"
       implicit val testAvroClassDecoder = specificAvroDeserializer[TestAvroClass](TestAvroClass.SCHEMA$)
 
       val producer = new KafkaProducer[String, TestAvroClass](Map(