Added a first stub to generate a custom KafkaProducer

Author: manub

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

@@ -2,19 +2,19 @@ package net.manub.embeddedkafka
 
 import java.net.InetSocketAddress
 import java.util.Properties
-import java.util.concurrent.{TimeUnit, Executors}
+import java.util.concurrent.Executors
 
 import kafka.consumer.{Consumer, ConsumerConfig, Whitelist}
 import kafka.serializer.StringDecoder
 import kafka.server.{KafkaConfig, KafkaServer}
 import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
-import org.apache.kafka.common.serialization.StringSerializer
+import org.apache.kafka.common.serialization.{Serializer, StringSerializer}
 import org.apache.zookeeper.server.{ServerCnxnFactory, ZooKeeperServer}
 import org.scalatest.Suite
 
 import scala.collection.JavaConversions.mapAsJavaMap
-import scala.concurrent.duration._
 import scala.concurrent._
+import scala.concurrent.duration._
 import scala.language.postfixOps
 import scala.reflect.io.Directory
 import scala.util.Try
@@ -99,8 +99,8 @@ trait EmbeddedKafka {
     val messageStreams =
       consumer.createMessageStreamsByFilter(filter, keyDecoder = new StringDecoder, valueDecoder = new StringDecoder)
 
-    val messageFuture = Future {
-      messageStreams.headOption.getOrElse(throw new KafkaSpecException("Unable to find a message stream")).iterator().next().message()
+    val messageFuture = Future { messageStreams.headOption
+      .getOrElse(throw new KafkaSpecException("Unable to find a message stream")).iterator().next().message()
     }
 
     try {
@@ -110,6 +110,23 @@ trait EmbeddedKafka {
     }
   }
 
+  def aKafkaProducerThat(): KafkaProducerConfiguration = {
+    new KafkaProducerConfiguration()
+  }
+
+  sealed class KafkaProducerConfiguration {
+
+    def serializesValuesWith[T <: Serializer[_]](serializer: Class[T])(implicit config: EmbeddedKafkaConfig) = {
+      new KafkaProducer[String, T](Map(
+        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG       -> s"localhost:${config.kafkaPort}",
+        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG    -> classOf[StringSerializer].getName,
+        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG  -> serializer.getName,
+        ProducerConfig.METADATA_FETCH_TIMEOUT_CONFIG  -> 3000.toString,
+        ProducerConfig.RETRY_BACKOFF_MS_CONFIG        -> 1000.toString
+      ))
+    }
+  }
+
 
   private def startZooKeeper(zooKeeperPort: Int): ServerCnxnFactory = {
     val zkLogsDir = Directory.makeTemp("zookeeper-logs")

src/test/scala/net/manub/embeddedkafka/EmbeddedKafkaSpec.scala

@@ -1,5 +1,6 @@
 package net.manub.embeddedkafka
 
+import java.lang.Class
 import java.net.InetSocketAddress
 import java.util.Properties
 import java.util.concurrent.TimeoutException
@@ -11,7 +12,7 @@ import akka.testkit.{ImplicitSender, TestKit}
 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
+import org.apache.kafka.common.serialization.{ByteArraySerializer, StringSerializer}
 import org.scalatest.concurrent.{JavaFutures, ScalaFutures}
 import org.scalatest.exceptions.TestFailedException
 import org.scalatest.time.{Milliseconds, Seconds, Span}
@@ -22,6 +23,7 @@ import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
 import scala.concurrent.duration._
 import scala.language.postfixOps
+import scala.reflect.ClassTag
 
 class EmbeddedKafkaSpec
   extends TestKit(ActorSystem("embedded-kafka-spec")) with WordSpecLike with EmbeddedKafka with Matchers
@@ -189,6 +191,15 @@ class EmbeddedKafkaSpec
     }
   }
 
+  "the aKafkaProducerThat method" should {
+
+    "return a producer that encodes messages for the given encoder" in {
+        val producer = aKafkaProducerThat serializesValuesWith classOf[ByteArraySerializer]
+
+        producer.isInstanceOf[KafkaProducer[String, ByteArraySerializer]] shouldBe true
+    }
+  }
+
   lazy val consumerConfigForEmbeddedKafka: ConsumerConfig = {
     val props = new Properties()
     props.put("group.id", "test")