Introducing support for Kafka streams (#39)

* Moved embedded kafka to its own sub module. Changed root project to aggregation project in preparation for introducing kafka-streams subproject.

* Introduced kafka-streams module for testing Kafka Streams using Embedded Kafka.
Introduced EmbeddedKafkaStreams and EmbeddedKafkaStreamsAllInOne traits for easy testing.
Introduced Consumers trait for easy creation and consumption of test messages.

* Disabling verbose logging for tests to make it easier to trace failed tests in build.

Author: adamosloizou

README.md

@@ -94,3 +94,73 @@ It is possible to create producers for custom types in two ways:
 
 For more information about how to use the utility methods, you can either look at the Scaladocs or at the tests of this project.
 
+## Custom consumers
+
+Use the `Consumer` trait that easily creates consumers of arbitrary key-value types and manages their lifecycle (via a loaner pattern).
+* For basic String consumption use `Consumer.withStringConsumer { your code here }`.
+* For arbitrary key and value types, expose implicit `Deserializer`s for each type and use `Consumer.withConsumer { your code here }`.
+* If you just want to create a consumer and manage its lifecycle yourself then try `Consumer.newConsumer()`.
+
+
+## Easy message consumption
+With `ConsumerExtensions` you can turn a consumer to a Scala lazy Stream of key-value pairs and treat it as a collection for easy assertion.
+* Just import the extensions.
+* On any `KafkaConsumer` instance you can now do:
+ 
+```scala
+import net.manub.embeddedkafka.ConsumerExtensions._
+...
+consumer.consumeLazily("from-this-topic").take(3).toList should be (Seq(
+  "1" -> "one", 
+  "2" -> "two", 
+  "3" -> "three"
+)
+```
+
+
+# scalatest-embedded-kafka-streams
+
+A library that builds on top of `scalatest-embedded-kafka` to offer easy testing of [Kafka Streams](https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams) with ScalaTest.
+It uses Kafka Streams 0.10.0.1.
+It takes care of instantiating and starting your streams as well as closing them after running your test-case code.
+
+## How to use
+
+* In your `build.sbt` file add the following dependency: `"net.manub" %% "scalatest-embedded-kafka-streams" % "0.8.1" % "test"`
+* Have a look at the [example test](kafka-streams/src/test/scala/net/manub/embeddedkafka/streams/ExampleKafkaStreamsSpec.scala)
+* For most of the cases have your `Spec` extend the `EmbeddedKafkaStreamsAllInOne` trait. This offers both streams management and easy creation of consumers for asserting resulting messages in output/sink topics.
+* If you only want to use the streams management without the test consumers just have the `Spec` extend the `EmbeddedKafkaStreams` trait.
+* Use the `runStreamsWithStringConsumer` to:
+    * Create any topics that need to exist for the strems to operate (usually sources and sinks).
+    * Pass the Stream or Topology builder that will then be used to instantiate and start the Kafka Streams. This will be done while using the `withRunningKafka` closure internally so that your stream runs with an embedded Kafka and Zookeeper.
+    * Pass the `{code block}` that needs a running instance of your streams. This is where your actual test code will sit. You can publish messages to your source topics and consume messages from your sink topics that the Kafka Streams should have generated. This method also offers a pre-instantiated consumer that can read String keys and values.
+* For more flexibility, use `runStreams` and `withConsumer`. This allows you to create your own consumers of custom types as seen in the [example test](kafka-streams/src/test/scala/net/manub/embeddedkafka/streams/ExampleKafkaStreamsSpec.scala).
+
+```scala
+import net.manub.embeddedkafka.ConsumerExtensions._
+import org.apache.kafka.streams.kstream.KStreamBuilder
+import org.scalatest.{Matchers, WordSpec}
+
+class MySpec extends WordSpec with Matchers with EmbeddedKafkaStreamsAllInOne {
+  "my kafka stream" should {
+    "be easy to test" in {
+      val inputTopic = "input-topic"
+      val outputTopic = "output-topic"
+      // your code for building the stream goes here e.g.
+      val streamBuilder = new KStreamBuilder
+      streamBuilder.stream(inputTopic).to(outputTopic)
+      // tell the stream test
+      // 1. what topics need to be created before the stream starts
+      // 2. the builder to be used for initializing and starting the stream
+      runStreamsWithStringConsumer(
+        topicsToCreate = Seq(inputTopic, outputTopic),
+        builder = streamBuilder
+      ){ consumer =>
+        // your test code goes here
+        publishToKafka(inputTopic, key = "hello", message = "world")
+        consumer.consumeLazily(outputTopic).head should be ("hello" -> "world")
+      }
+    }
+  }
+}
+```

build.sbt

@@ -1,23 +1,27 @@
 import sbtrelease.Version
 
+val kafkaVersion = "0.10.0.1"
+
 val slf4jLog4jOrg = "org.slf4j"
 val slf4jLog4jArtifact = "slf4j-log4j12"
 
 lazy val commonSettings = Seq(
-  name := "scalatest-embedded-kafka",
   organization := "net.manub",
   scalaVersion := "2.11.8",
   crossScalaVersions := Seq("2.10.6", "2.11.8"),
   homepage := Some(url("https://github.com/manub/scalatest-embedded-kafka")),
   parallelExecution in Test := false,
-  libraryDependencies ++= Seq(
-    "org.scalatest" %% "scalatest" % "3.0.0",
-    "org.apache.kafka" %% "kafka" % "0.10.0.1" exclude(slf4jLog4jOrg, slf4jLog4jArtifact),
-    "org.apache.zookeeper" % "zookeeper" % "3.4.7" exclude(slf4jLog4jOrg, slf4jLog4jArtifact),
-    "org.apache.avro" % "avro" % "1.7.7" exclude(slf4jLog4jOrg, slf4jLog4jArtifact),
-    "com.typesafe.akka" %% "akka-actor" % "2.3.14" % Test,
-    "com.typesafe.akka" %% "akka-testkit" % "2.3.14" % Test
-  )
+  logBuffered in Test := false,
+  fork in Test := true
+)
+
+lazy val commonLibrarySettings = libraryDependencies ++= Seq(
+  "org.scalatest" %% "scalatest" % "3.0.0",
+  "org.apache.kafka" %% "kafka" % kafkaVersion exclude(slf4jLog4jOrg, slf4jLog4jArtifact),
+  "org.apache.zookeeper" % "zookeeper" % "3.4.7" exclude(slf4jLog4jOrg, slf4jLog4jArtifact),
+  "org.apache.avro" % "avro" % "1.7.7" exclude(slf4jLog4jOrg, slf4jLog4jArtifact),
+  "com.typesafe.akka" %% "akka-actor" % "2.3.14" % Test,
+  "com.typesafe.akka" %% "akka-testkit" % "2.3.14" % Test
 )
 
 lazy val publishSettings = Seq(
@@ -47,6 +51,25 @@ lazy val releaseSettings = Seq(
 )
 
 lazy val root = (project in file("."))
+    .settings(name := "scalatest-embedded-kafka-root")
+    .settings(commonSettings: _*)
+    .aggregate(embeddedKafka, kafkaStreams)
+
+
+lazy val embeddedKafka = (project in file("embedded-kafka"))
+  .settings(name := "scalatest-embedded-kafka")
+  .settings(publishSettings: _*)
+  .settings(commonSettings: _*)
+  .settings(commonLibrarySettings)
+  .settings(releaseSettings: _*)
+
+lazy val kafkaStreams = (project in file("kafka-streams"))
+  .settings(name := "scalatest-embedded-kafka-streams")
   .settings(publishSettings: _*)
   .settings(commonSettings: _*)
+  .settings(commonLibrarySettings)
   .settings(releaseSettings: _*)
+  .settings(libraryDependencies ++= Seq(
+    "org.apache.kafka" % "kafka-streams" % kafkaVersion exclude(slf4jLog4jOrg, slf4jLog4jArtifact)
+  ))
+  .dependsOn(embeddedKafka)

src/main/scala/net/manub/embeddedkafka/Codecs.scala renamed to embedded-kafka/src/main/scala/net/manub/embeddedkafka/Codecs.scala

@@ -0,0 +1,51 @@
+package net.manub.embeddedkafka
+
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.common.KafkaException
+import org.apache.log4j.Logger
+
+import scala.util.Try
+
+/** Method extensions for Kafka's [[KafkaConsumer]] API allowing easy testing.*/
+object ConsumerExtensions {
+  val MaximumAttempts = 3
+  implicit class ConsumerOps[K, V](val consumer: KafkaConsumer[K, V]) {
+
+    private val logger = Logger.getLogger(classOf[ConsumerOps[K, V]])
+
+    /** Consume messages from a given topic and return them as a lazily evaluated Scala Stream.
+      * Depending on how many messages are taken from the Scala Stream it will try up to 3 times
+      * to consume batches from the given topic, until it reaches the number of desired messages or
+      * return otherwise.
+      *
+      * @param topic the topic from which to consume messages
+      * @return the stream of consumed messages that you can do `.take(n: Int).toList`
+      *         to evaluate the requested number of messages.
+      */
+    def consumeLazily(topic: String): Stream[(K, V)] = {
+      val attempts = 1 to MaximumAttempts
+      attempts.toStream.flatMap { attempt =>
+        val batch: Seq[(K, V)] = getNextBatch(topic)
+        logger.debug(s"----> Batch $attempt ($topic) | ${batch.mkString("|")}")
+        batch
+      }
+    }
+
+    /** Get the next batch of messages from Kafka.
+      *
+      * @param topic the topic to consume
+      * @return the next batch of messages
+      */
+    def getNextBatch(topic: String): Seq[(K, V)] = Try {
+      import scala.collection.JavaConversions._
+      consumer.subscribe(List(topic))
+      consumer.partitionsFor(topic)
+      val records = consumer.poll(2000)
+      // use toList to force eager evaluation. toSeq is lazy
+      records.iterator().toList.map(r => r.key -> r.value)
+    }.recover {
+      case ex: KafkaException => throw new KafkaUnavailableException(ex)
+    }.get
+  }
+}
+

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

@@ -0,0 +1,59 @@
+package net.manub.embeddedkafka
+
+import java.util.Properties
+
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.common.serialization.Deserializer
+
+/** Utility trait for easily creating Kafka consumers and accessing their consumed messages. */
+trait Consumers {
+  /** Loaner pattern that allows running a code block with a newly created consumer.
+    * The consumer's lifecycle will be automatically handled and closed at the end of the
+    * given code block.
+    *
+    * @param block the code block to be executed with the instantiated consumer
+    *              passed as an argument
+    * @tparam K the type of the consumer's Key
+    * @tparam V the type of the consumer's Value
+    * @tparam T the type of the block's returning result
+    * @return the result of the executed block
+    */
+  def withConsumer[K: Deserializer, V: Deserializer, T](block: KafkaConsumer[K, V] => T)
+                                                       (implicit config: EmbeddedKafkaConfig): T = {
+    val consumer = newConsumer[K, V]()
+    try {
+      val result = block(consumer)
+      result
+    } finally {
+      consumer.close()
+    }
+  }
+
+  /** Convenience alternative to `withConsumer` that offers a consumer for String keys and values.
+    *
+    * @param block the block to be executed with the consumer
+    * @tparam T the type of the result of the code block
+    * @return the code block result
+    */
+  def withStringConsumer[T](block: KafkaConsumer[String, String] => T)
+                           (implicit config: EmbeddedKafkaConfig): T = {
+    import net.manub.embeddedkafka.Codecs.stringDeserializer
+    withConsumer(block)
+  }
+
+  /** Create a new Kafka consumer.
+    *
+    * @tparam K the type of the consumer's Key
+    * @tparam V the type of the consumer's Value
+    * @return the new consumer
+    */
+  def newConsumer[K: Deserializer, V: Deserializer]()
+                                                   (implicit config: EmbeddedKafkaConfig): KafkaConsumer[K, V] = {
+    val props = new Properties()
+    props.put("group.id", UUIDs.newUuid().toString)
+    props.put("bootstrap.servers", s"localhost:${config.kafkaPort}")
+    props.put("auto.offset.reset", "earliest")
+
+    new KafkaConsumer[K, V](props, implicitly[Deserializer[K]], implicitly[Deserializer[V]])
+  }
+}

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

@@ -0,0 +1,14 @@
+package net.manub.embeddedkafka
+
+import java.util.UUID
+
+/** Utility object for creating unique test IDs.
+  * Useful for separating IDs and directories across test cases.
+  */
+object UUIDs {
+  /** Create a new unique ID.
+    *
+    * @return the unique ID
+    */
+  def newUuid(): UUID = UUID.randomUUID()
+}