Using headOption instead of head for accessing first element in a list.

Author: manub

src/main/scala/net/manub/embeddedkafka/EmbeddedKafka.scala

@@ -99,7 +99,9 @@ trait EmbeddedKafka {
     val messageStreams =
       consumer.createMessageStreamsByFilter(filter, keyDecoder = new StringDecoder, valueDecoder = new StringDecoder)
 
-    val messageFuture = Future { messageStreams.head.iterator().next().message() }
+    val messageFuture = Future {
+      messageStreams.headOption.getOrElse(throw new KafkaSpecException("Unable to find a message stream")).iterator().next().message()
+    }
 
     try {
       Await.result(messageFuture, 3 seconds)

src/main/scala/net/manub/embeddedkafka/KafkaUnavailableException.scala

@@ -0,0 +1,5 @@
+package net.manub.embeddedkafka
+
+class KafkaUnavailableException extends RuntimeException
+
+class KafkaSpecException(msg: String) extends RuntimeException(msg)

src/main/scala/net/manub/embeddedkafka/exceptions.scala

@@ -8,7 +8,7 @@ import akka.actor.{Actor, ActorRef, ActorSystem, Props}
 import akka.io.Tcp._
 import akka.io.{IO, Tcp}
 import akka.testkit.{ImplicitSender, TestKit}
-import kafka.consumer.{ConsumerConfig, Consumer, Whitelist}
+import kafka.consumer.{Consumer, ConsumerConfig, Whitelist}
 import kafka.serializer.StringDecoder
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
 import org.apache.kafka.common.serialization.StringSerializer
@@ -124,7 +124,12 @@ class EmbeddedKafkaSpec
         val messageStreams = consumer.createMessageStreamsByFilter(filter, 1, stringDecoder, stringDecoder)
 
         val eventualMessage = Future {
-          messageStreams.head.iterator().next().message()
+          messageStreams
+            .headOption
+            .getOrElse(throw new RuntimeException("Unable to retrieve message streams"))
+            .iterator()
+            .next()
+            .message()
         }
 
         whenReady(eventualMessage) { msg =>