Kafka 2.0.0 (#154)

Author: NeQuissimus

.travis.yml

@@ -1,6 +1,6 @@
 language: scala
 
-jdk: 
+jdk:
   - oraclejdk8
 
 scala:
@@ -19,6 +19,5 @@ before_cache:
   - find $HOME/.ivy2/cache -name "ivydata-*.properties" -print -delete
   - find $HOME/.sbt        -name "*.lock"               -print -delete
 
-script: 
+script:
   - travis_retry sbt ++$TRAVIS_SCALA_VERSION test
-

build.sbt

@@ -2,8 +2,8 @@ import sbtrelease.Version
 
 parallelExecution in ThisBuild := false
 
-val kafkaVersion = "1.1.1"
-val confluentVersion = "4.1.1"
+val kafkaVersion = "2.0.0"
+val confluentVersion = "5.0.0"
 val akkaVersion = "2.5.14"
 
 lazy val commonSettings = Seq(
@@ -22,6 +22,7 @@ lazy val commonSettings = Seq(
 lazy val commonLibrarySettings = libraryDependencies ++= Seq(
   "org.apache.avro" % "avro" % "1.8.2",
   "org.apache.kafka" %% "kafka" % kafkaVersion,
+  "org.slf4j" % "slf4j-log4j12" % "1.7.25" % Test,
   "org.scalatest" %% "scalatest" % "3.0.5" % Test,
   "com.typesafe.akka" %% "akka-actor" % akkaVersion % Test,
   "com.typesafe.akka" %% "akka-testkit" % akkaVersion % Test
@@ -53,6 +54,13 @@ lazy val releaseSettings = Seq(
   releaseCrossBuild := true
 )
 
+// https://github.com/sbt/sbt/issues/3618
+// [error] (kafkaStreams / update) sbt.librarymanagement.ResolveException: download failed: javax.ws.rs#javax.ws.rs-api;2.1!javax.ws.rs-api.${packaging.type}
+val workaround = {
+  sys.props += "packaging.type" -> "jar"
+  ()
+}
+
 lazy val root = (project in file("."))
   .settings(name := "scalatest-embedded-kafka-root")
   .settings(commonSettings: _*)

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

@@ -1,19 +1,14 @@
 package net.manub.embeddedkafka
 
-import kafka.serializer._
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.common.serialization._
 
 /** useful encoders/serializers, decoders/deserializers and [[ConsumerRecord]] decoders**/
 object Codecs {
-  implicit val stringEncoder: Encoder[String] = new StringEncoder()
-  implicit val nullEncoder: Encoder[Array[Byte]] = new DefaultEncoder()
   implicit val stringSerializer: Serializer[String] = new StringSerializer()
   implicit val nullSerializer: Serializer[Array[Byte]] =
     new ByteArraySerializer()
 
-  implicit val stringDecoder: Decoder[String] = new StringDecoder()
-  implicit val nullDecoder: Decoder[Array[Byte]] = new DefaultDecoder()
   implicit val stringDeserializer: Deserializer[String] =
     new StringDeserializer()
   implicit val nullDeserializer: Deserializer[Array[Byte]] =

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

@@ -48,7 +48,7 @@ object ConsumerExtensions {
         import scala.collection.JavaConverters._
         consumer.subscribe(topics.asJava)
         topics.foreach(consumer.partitionsFor)
-        val records = consumer.poll(poll)
+        val records = consumer.poll(java.time.Duration.ofMillis(poll))
         // use toList to force eager evaluation. toSeq is lazy
         records.iterator().asScala.toList.map(decoder(_))
       }.recover {

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

@@ -588,7 +588,7 @@ private[embeddedkafka] trait EmbeddedKafkaSupport[C <: EmbeddedKafkaConfig] {
       topics.foreach(consumer.partitionsFor)
 
       while (messagesRead < number && System.nanoTime < timeoutNanoTime) {
-        val records = consumer.poll(1000)
+        val records = consumer.poll(java.time.Duration.ofMillis(1000))
         val recordIter = records.iterator()
         if (resetTimeoutOnEachMessage && recordIter.hasNext) {
           timeoutNanoTime = System.nanoTime + timeout.toNanos

embedded-kafka/src/main/scala/net/manub/embeddedkafka/avro/avroMarshallers.scala

@@ -2,7 +2,6 @@ package net.manub.embeddedkafka.avro
 
 import java.io.ByteArrayOutputStream
 
-import kafka.serializer.{Decoder, Encoder}
 import kafka.utils.VerifiableProperties
 import org.apache.avro.Schema
 import org.apache.avro.io._
@@ -13,58 +12,39 @@ import org.apache.avro.specific.{
 }
 import org.apache.kafka.common.serialization.{Deserializer, Serializer}
 
-class KafkaAvroDecoder[T <: SpecificRecord](schema: Schema,
-                                            props: VerifiableProperties = null)
-    extends Decoder[T] {
-  private val NoInstanceReuse = null.asInstanceOf[T]
-  private val NoDecoderReuse = null.asInstanceOf[BinaryDecoder]
+class KafkaAvroDeserializer[T <: SpecificRecord](schema: Schema)
+    extends Deserializer[T]
+    with NoOpConfiguration
+    with NoOpClose {
+
   private val reader = new SpecificDatumReader[T](schema)
 
-  override def fromBytes(bytes: Array[Byte]): T = {
-    val decoder = DecoderFactory.get().binaryDecoder(bytes, NoDecoderReuse)
-    reader.read(NoInstanceReuse, decoder)
+  override def deserialize(topic: String, data: Array[Byte]): T = {
+    val decoder = DecoderFactory.get().binaryDecoder(data, null)
+    reader.read(null.asInstanceOf[T], decoder)
   }
 }
 
-class KafkaAvroEncoder[T <: SpecificRecord](props: VerifiableProperties = null)
-    extends Encoder[T] {
-  private val NoEncoderReuse = null.asInstanceOf[BinaryEncoder]
+class KafkaAvroSerializer[T <: SpecificRecord]()
+    extends Serializer[T]
+    with NoOpConfiguration
+    with NoOpClose {
 
-  override def toBytes(nullableData: T): Array[Byte] = {
+  private def toBytes(nullableData: T): Array[Byte] =
     Option(nullableData).fold[Array[Byte]](null) { data =>
       val writer: DatumWriter[T] = new SpecificDatumWriter[T](data.getSchema)
       val out = new ByteArrayOutputStream()
-      val encoder = EncoderFactory.get.binaryEncoder(out, NoEncoderReuse)
+      val encoder = EncoderFactory.get.binaryEncoder(out, null)
 
       writer.write(data, encoder)
       encoder.flush()
       out.close()
 
       out.toByteArray
     }
-  }
-}
-
-class KafkaAvroDeserializer[T <: SpecificRecord](schema: Schema)
-    extends Deserializer[T]
-    with NoOpConfiguration
-    with NoOpClose {
-
-  private val decoder = new KafkaAvroDecoder[T](schema = schema)
-
-  override def deserialize(topic: String, data: Array[Byte]): T =
-    decoder.fromBytes(data)
-}
-
-class KafkaAvroSerializer[T <: SpecificRecord]()
-    extends Serializer[T]
-    with NoOpConfiguration
-    with NoOpClose {
-
-  private val encoder = new KafkaAvroEncoder[T]()
 
   override def serialize(topic: String, data: T): Array[Byte] =
-    encoder.toBytes(data)
+    toBytes(data)
 }
 
 sealed trait NoOpConfiguration {

embedded-kafka/src/main/scala/net/manub/embeddedkafka/avro/package.scala

@@ -1,24 +1,15 @@
 package net.manub.embeddedkafka
 
-import kafka.serializer.{Decoder, Encoder}
 import kafka.utils.VerifiableProperties
 import org.apache.avro.Schema
 import org.apache.avro.specific.SpecificRecord
 import org.apache.kafka.common.serialization.{Deserializer, Serializer}
 
 package object avro {
-
   implicit def specificAvroSerializer[T <: SpecificRecord]: Serializer[T] =
     new KafkaAvroSerializer[T]
-  implicit def specificAvroEncoder[T <: SpecificRecord]: Encoder[T] =
-    new KafkaAvroEncoder[T]
 
   def specificAvroDeserializer[T <: SpecificRecord](
       schema: Schema): Deserializer[T] =
     new KafkaAvroDeserializer[T](schema)
-
-  def specificAvroDecoder[T <: SpecificRecord](schema: Schema,
-                                               props: VerifiableProperties =
-                                                 null): Decoder[T] =
-    new KafkaAvroDecoder[T](schema, props)
 }

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

@@ -30,11 +30,13 @@ class ConsumerExtensionsSpec
           .empty[TopicPartition, java.util.List[ConsumerRecord[String, String]]]
           .asJava)
 
-      when(consumer.poll(retryConf.poll)).thenReturn(consumerRecords)
+      when(consumer.poll(java.time.Duration.ofMillis(retryConf.poll)))
+        .thenReturn(consumerRecords)
 
       consumer.consumeLazily[String]("topic")
 
-      verify(consumer, times(retryConf.maximumAttempts)).poll(retryConf.poll)
+      verify(consumer, times(retryConf.maximumAttempts))
+        .poll(java.time.Duration.ofMillis(retryConf.poll))
     }
 
     "not retry to get messages with the configured maximum number of attempts when poll succeeds" in {
@@ -48,11 +50,12 @@ class ConsumerExtensionsSpec
           new TopicPartition("topic", 1) -> List(consumerRecord).asJava).asJava
       )
 
-      when(consumer.poll(retryConf.poll)).thenReturn(consumerRecords)
+      when(consumer.poll(java.time.Duration.ofMillis(retryConf.poll)))
+        .thenReturn(consumerRecords)
 
       consumer.consumeLazily[String]("topic")
 
-      verify(consumer).poll(retryConf.poll)
+      verify(consumer).poll(java.time.Duration.ofMillis(retryConf.poll))
     }
 
     "poll to get messages with the configured poll timeout" in {
@@ -65,11 +68,12 @@ class ConsumerExtensionsSpec
           .empty[TopicPartition, java.util.List[ConsumerRecord[String, String]]]
           .asJava)
 
-      when(consumer.poll(retryConf.poll)).thenReturn(consumerRecords)
+      when(consumer.poll(java.time.Duration.ofMillis(retryConf.poll)))
+        .thenReturn(consumerRecords)
 
       consumer.consumeLazily[String]("topic")
 
-      verify(consumer).poll(retryConf.poll)
+      verify(consumer).poll(java.time.Duration.ofMillis(retryConf.poll))
     }
   }
 

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

@@ -51,7 +51,8 @@ class EmbeddedKafkaMethodsSpec
       val consumer = kafkaConsumer
       consumer.subscribe(List(topic).asJava)
 
-      val records = consumer.poll(consumerPollTimeout)
+      val records =
+        consumer.poll(java.time.Duration.ofMillis(consumerPollTimeout))
 
       records.iterator().hasNext shouldBe true
       val record = records.iterator().next()
@@ -77,7 +78,8 @@ class EmbeddedKafkaMethodsSpec
       val consumer = kafkaConsumer
       consumer.subscribe(List(topic).asJava)
 
-      val records = consumer.poll(consumerPollTimeout)
+      val records =
+        consumer.poll(java.time.Duration.ofMillis(consumerPollTimeout))
 
       records.iterator().hasNext shouldBe true
       val record = records.iterator().next()
@@ -102,7 +104,8 @@ class EmbeddedKafkaMethodsSpec
       val consumer = kafkaConsumer
       consumer.subscribe(List(topic).asJava)
 
-      val records = consumer.poll(consumerPollTimeout)
+      val records =
+        consumer.poll(java.time.Duration.ofMillis(consumerPollTimeout))
 
       records.iterator().hasNext shouldBe true
       val record = records.iterator().next()
@@ -129,7 +132,9 @@ class EmbeddedKafkaMethodsSpec
       val consumer = kafkaConsumer
       consumer.subscribe(List(topic).asJava)
 
-      val records = consumer.poll(consumerPollTimeout).iterator()
+      val records = consumer
+        .poll(java.time.Duration.ofMillis(consumerPollTimeout))
+        .iterator()
 
       records.hasNext shouldBe true
 

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

@@ -110,7 +110,8 @@ class EmbeddedKafkaObjectSpec extends EmbeddedKafkaSpecSupport {
           kafkaConsumer(someOtherConfig, deserializer, deserializer)
         anotherConsumer.subscribe(List(topic).asJava)
 
-        val moreRecords = anotherConsumer.poll(consumerPollTimeout)
+        val moreRecords =
+          anotherConsumer.poll(java.time.Duration.ofMillis(consumerPollTimeout))
         moreRecords.count shouldBe 1
 
         val someOtherRecord = moreRecords.iterator().next