Allow consumeFirstStringMessageFrom to be called multiple times

In older versions of scalatest-embedded-kafka you were able to call
consumeFirstStringMessageFrom multiple times, with each invocation
causing the next message in a topic to be consumed. In the latest
version this is 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. We fix this by disabling auto-commit 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
@@ -236,6 +236,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", "false")
 
     val consumer =
       new KafkaConsumer[String, T](props, new StringDeserializer, deserializer)
@@ -248,7 +249,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/test/scala/net/manub/embeddedkafka/EmbeddedKafkaMethodsSpec.scala

@@ -44,6 +44,9 @@ class EmbeddedKafkaMethodsSpec extends EmbeddedKafkaSpecSupport with EmbeddedKaf
 
       consumer.close()
 
+      // Commit message in embedded-kafka-spec group so we don't consume it
+      // again in the consumeFirstStringMessageFrom tests below.
+      consumeFirstStringMessageFrom(topic)
     }
 
     "publish synchronously a String message with String key to Kafka" in {
@@ -67,6 +70,10 @@ class EmbeddedKafkaMethodsSpec extends EmbeddedKafkaSpecSupport with EmbeddedKaf
 
 
       consumer.close()
+
+      // Commit message in embedded-kafka-spec group so we don't consume it
+      // again in the consumeFirstStringMessageFrom tests below.
+      consumeFirstStringMessageFrom(topic)
     }
   }
 
@@ -126,6 +133,31 @@ 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 = "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"