diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/DefaultSerdes.scala b/src/main/scala/com/lightbend/kafka/scala/streams/DefaultSerdes.scala deleted file mode 100644 index 1996a982f..000000000 --- a/src/main/scala/com/lightbend/kafka/scala/streams/DefaultSerdes.scala +++ /dev/null @@ -1,20 +0,0 @@ -/** - * Copyright (C) 2018 Lightbend Inc. - * Copyright 2017-2018 Alexis Seigneurin. - */ -package com.lightbend.kafka.scala.streams - -import org.apache.kafka.common.serialization.{Serde, Serdes} - -/** - * Implicit values for default serdes - */ -object DefaultSerdes { - implicit val stringSerde: Serde[String] = Serdes.String() - implicit val longSerde: Serde[Long] = Serdes.Long().asInstanceOf[Serde[Long]] - implicit val byteArraySerde: Serde[Array[Byte]] = Serdes.ByteArray() - implicit val bytesSerde: Serde[org.apache.kafka.common.utils.Bytes] = Serdes.Bytes() - implicit val floatSerde: Serde[Float] = Serdes.Float().asInstanceOf[Serde[Float]] - implicit val doubleSerde: Serde[Double] = Serdes.Double().asInstanceOf[Serde[Double]] - implicit val integerSerde: Serde[Int] = Serdes.Integer().asInstanceOf[Serde[Int]] -} diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala b/src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala index d9a087db7..7a673ac9a 100644 --- a/src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala +++ b/src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala @@ -7,7 +7,6 @@ package com.lightbend.kafka.scala.streams import org.apache.kafka.streams.kstream._ import org.apache.kafka.streams.{Consumed, KeyValue} import org.apache.kafka.common.serialization.Serde - import scala.language.implicitConversions /** diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/ScalaSerde.scala b/src/main/scala/com/lightbend/kafka/scala/streams/ScalaSerde.scala deleted file mode 100644 index 2c9f73c46..000000000 --- a/src/main/scala/com/lightbend/kafka/scala/streams/ScalaSerde.scala +++ /dev/null @@ -1,47 +0,0 @@ -// adopted from Openshine implementation -package com.lightbend.kafka.scala.streams - -import org.apache.kafka.common.serialization.{Serde, Deserializer => JDeserializer, Serializer => JSerializer} - -trait ScalaSerde[T] extends Serde[T] { - override def deserializer(): JDeserializer[T] - - override def serializer(): JSerializer[T] - - override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = () - - override def close(): Unit = () -} - -trait StatelessScalaSerde[T >: Null] extends Serde[T] with ScalaSerde[T] { - def serialize(data: T): Array[Byte] - def deserialize(data: Array[Byte]): Option[T] - - override def deserializer(): Deserializer[T] = - (data: Array[Byte]) => deserialize(data) - - override def serializer(): Serializer[T] = - (data: T) => serialize(data) -} - -trait Deserializer[T >: Null] extends JDeserializer[T] { - override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = () - - override def close(): Unit = () - - override def deserialize(topic: String, data: Array[Byte]): T = - Option(data).flatMap(deserialize).orNull - - def deserialize(data: Array[Byte]): Option[T] -} - -trait Serializer[T] extends JSerializer[T] { - override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = () - - override def close(): Unit = () - - override def serialize(topic: String, data: T): Array[Byte] = - Option(data).map(serialize).orNull - - def serialize(data: T): Array[Byte] -} diff --git a/src/main/scala/com/lightbend/kafka/scala/streams/Serdes.scala b/src/main/scala/com/lightbend/kafka/scala/streams/Serdes.scala new file mode 100644 index 000000000..8148a5384 --- /dev/null +++ b/src/main/scala/com/lightbend/kafka/scala/streams/Serdes.scala @@ -0,0 +1,104 @@ +/** + * Copyright (C) 2018 Lightbend Inc. + * Copyright 2017-2018 Alexis Seigneurin. + */ +package com.lightbend.kafka.scala.streams + +import java.util +import org.apache.kafka.common.serialization.{ + Serde, + Serdes => JavaSerdes, + Deserializer => JavaDeserializer, + Serializer => JavaSerializer +} + +trait Serdes { + implicit val string: Serde[String] = JavaSerdes.String() + implicit val long: Serde[Long] = JavaSerdes.Long().asInstanceOf[Serde[Long]] + implicit val javaLong: Serde[java.lang.Long] = JavaSerdes.Long() + implicit val byteArray: Serde[Array[Byte]] = JavaSerdes.ByteArray() + implicit val bytes: Serde[org.apache.kafka.common.utils.Bytes] = JavaSerdes.Bytes() + implicit val float: Serde[Float] = JavaSerdes.Float().asInstanceOf[Serde[Float]] + implicit val javaFloat: Serde[java.lang.Float] = JavaSerdes.Float() + implicit val double: Serde[Double] = JavaSerdes.Double().asInstanceOf[Serde[Double]] + implicit val javaDouble: Serde[java.lang.Double] = JavaSerdes.Double() + implicit val integer: Serde[Int] = JavaSerdes.Integer().asInstanceOf[Serde[Int]] + implicit val javaInteger: Serde[java.lang.Integer] = JavaSerdes.Integer() + + def fromFn[T >: Null](serializer: T => Array[Byte], deserializer: Array[Byte] => Option[T]): Serde[T] = + JavaSerdes.serdeFrom( + new JavaSerializer[T] { + override def serialize(topic: String, data: T): Array[Byte] = serializer(data) + override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = () + override def close(): Unit = () + }, + new JavaDeserializer[T] { + override def deserialize(topic: String, data: Array[Byte]): T = deserializer(data).orNull + override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = () + override def close(): Unit = () + } + ) + + def fromFn[T >: Null](serializer: (String, T) => Array[Byte], + deserializer: (String, Array[Byte]) => Option[T]): Serde[T] = + JavaSerdes.serdeFrom( + new JavaSerializer[T] { + override def serialize(topic: String, data: T): Array[Byte] = serializer(topic, data) + override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = () + override def close(): Unit = () + }, + new JavaDeserializer[T] { + override def deserialize(topic: String, data: Array[Byte]): T = deserializer(topic, data).orNull + override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = () + override def close(): Unit = () + } + ) +} + +/** + * Implicit values for default serdes + */ +object Serdes extends Serdes + +@deprecated("Use com.lightbend.kafka.scala.streams.Serdes instead", "") +object DefaultSerdes extends Serdes + +@deprecated("Use com.lightbend.kafka.scala.streams.Serdes.fromFn instead", "") +trait StatelessSerde[T >: Null] extends Serde[T] { + def serialize(data: T): Array[Byte] + def deserialize(data: Array[Byte]): Option[T] + + override def deserializer(): Deserializer[T] = + (data: Array[Byte]) => deserialize(data) + + override def serializer(): Serializer[T] = + (data: T) => serialize(data) + + override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = () + + override def close(): Unit = () +} + +@deprecated("Use org.apache.kafka.common.serialization.Deserializer instead", "") +trait Deserializer[T >: Null] extends JavaDeserializer[T] { + override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = () + + override def close(): Unit = () + + override def deserialize(topic: String, data: Array[Byte]): T = + Option(data).flatMap(deserialize).orNull + + def deserialize(data: Array[Byte]): Option[T] +} + +@deprecated("Use org.apache.kafka.common.serialization.Serializer instead", "") +trait Serializer[T] extends JavaSerializer[T] { + override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = () + + override def close(): Unit = () + + override def serialize(topic: String, data: T): Array[Byte] = + Option(data).map(serialize).orNull + + def serialize(data: T): Array[Byte] +} diff --git a/src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsMergeTest.scala b/src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsMergeTest.scala index bef2f5e76..3f38ac23a 100644 --- a/src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsMergeTest.scala +++ b/src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsMergeTest.scala @@ -33,7 +33,7 @@ object KafkaStreamsMergeTest extends TestSuite[KafkaLocalServer] with WordCountM // // Step 1: Configure and start the processor topology. // - import DefaultSerdes._ + import Serdes._ val streamsConfiguration = new Properties() streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, s"wordcount-${scala.util.Random.nextInt(100)}") diff --git a/src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsTest.scala b/src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsTest.scala index 4c1a23277..a28a78a0f 100644 --- a/src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsTest.scala +++ b/src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsTest.scala @@ -32,7 +32,7 @@ object KafkaStreamsTest extends TestSuite[KafkaLocalServer] with WordCountTestDa // // Step 1: Configure and start the processor topology. // - import DefaultSerdes._ + import Serdes._ val streamsConfiguration = new Properties() streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, s"wordcount-${scala.util.Random.nextInt(100)}") diff --git a/src/test/scala/com/lightbend/kafka/scala/streams/ProbabilisticCountingScalaIntegrationTest.scala b/src/test/scala/com/lightbend/kafka/scala/streams/ProbabilisticCountingScalaIntegrationTest.scala index 6441d1b78..fe653dbd4 100644 --- a/src/test/scala/com/lightbend/kafka/scala/streams/ProbabilisticCountingScalaIntegrationTest.scala +++ b/src/test/scala/com/lightbend/kafka/scala/streams/ProbabilisticCountingScalaIntegrationTest.scala @@ -90,8 +90,8 @@ object ProbabilisticCountingScalaIntegrationTest p.put(StreamsConfig.APPLICATION_ID_CONFIG, s"probabilistic-counting-scala-integration-test-${scala.util.Random.nextInt(100)}") p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers) - p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.ByteArray.getClass.getName) - p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String.getClass.getName) + p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.byteArray.getClass.getName) + p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.string.getClass.getName) p.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "10000") p.put(StreamsConfig.STATE_DIR_CONFIG, localStateDir) p @@ -107,7 +107,7 @@ object ProbabilisticCountingScalaIntegrationTest cfg.put("segment.bytes", segmentSizeBytes) cfg } - new CMSStoreBuilder[String](cmsStoreName, Serdes.String()) + new CMSStoreBuilder[String](cmsStoreName, Serdes.string) .withLoggingEnabled(changelogConfig) } builder.addStateStore(cmsStoreBuilder) @@ -137,9 +137,9 @@ object ProbabilisticCountingScalaIntegrationTest override def close(): Unit = {} } - implicit val stringSerde: Serde[String] = Serdes.String() - implicit val byteArraySerde: Serde[Array[Byte]] = Serdes.ByteArray() - implicit val longSerde: Serde[Long] = Serdes.Long().asInstanceOf[Serde[Long]] + implicit val stringSerde: Serde[String] = Serdes.string + implicit val byteArraySerde: Serde[Array[Byte]] = Serdes.byteArray + implicit val longSerde: Serde[Long] = Serdes.long // Read the input from Kafka. val textLines: KStreamS[Array[Byte], String] = builder.stream(inputTopic) diff --git a/src/test/scala/com/lightbend/kafka/scala/streams/PunctuateTest.scala b/src/test/scala/com/lightbend/kafka/scala/streams/PunctuateTest.scala index d0510c536..86d026ee8 100644 --- a/src/test/scala/com/lightbend/kafka/scala/streams/PunctuateTest.scala +++ b/src/test/scala/com/lightbend/kafka/scala/streams/PunctuateTest.scala @@ -41,8 +41,8 @@ object PunctuateTest extends TestSuite[KafkaLocalServer] with PunctuateTestData streamsConfiguration.put(StreamsConfig.CLIENT_ID_CONFIG, "punctuategroup") streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers) - streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()) - streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()) + streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.string.getClass.getName) + streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.string.getClass.getName) val topology = new Topology // Data input streams diff --git a/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala b/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala index 5b08474e3..ed1ff131f 100644 --- a/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala +++ b/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala @@ -66,7 +66,7 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdes // DefaultSerdes brings into scope implicit serdes (mostly for primitives) that will set up all Serialized, Produced, // Consumed and Joined instances. So all APIs below that accept Serialized, Produced, Consumed or Joined will // get these instances automatically - import DefaultSerdes._ + import Serdes._ // we don't have any serde declared as part of configuration. Even if they are declared here, the // Scala APIs will ignore them. But it's possible to declare serdes here and use them through diff --git a/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro.scala b/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro.scala index e0279c994..af47f912c 100644 --- a/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro.scala +++ b/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro.scala @@ -36,28 +36,22 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro case class UserClicks(clicks: Long) - // adopted from Openshine implementation - class AvroSerde[T >: Null: SchemaFor: FromRecord: ToRecord] extends StatelessScalaSerde[T] { - - override def serialize(data: T): Array[Byte] = { + /** Our implicit Serde implementation for the values we want to serialize + * as avro + */ + implicit def avroSerde[T >: Null: SchemaFor: FromRecord: ToRecord]: Serde[T] = Serdes.fromFn( + { data: T => val baos = new ByteArrayOutputStream() val output = AvroOutputStream.binary[T](baos) output.write(data) output.close() baos.toByteArray - } - - override def deserialize(data: Array[Byte]): Option[T] = { + }, { data => val in = new ByteArrayInputStream(data) val input = AvroInputStream.binary[T](in) input.iterator.toSeq.headOption } - } - - /** Our implicit Serde implementation for the values we want to serialize - * as avro - */ - implicit val userClicksSerde: Serde[UserClicks] = new AvroSerde + ) /** * End-to-end integration test that demonstrates how to perform a join @@ -95,7 +89,7 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro // DefaultSerdes brings into scope implicit serdes (mostly for primitives) that will set up all Serialized, Produced, // Consumed and Joined instances. So all APIs below that accept Serialized, Produced, Consumed or Joined will // get these instances automatically - import DefaultSerdes._ + import Serdes._ // // Step 1: Configure and start the processor topology. @@ -147,11 +141,11 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro println(s"Stream terminated because of uncaught exception .. Shutting " + s"down app", e) - e.printStackTrace + e.printStackTrace() val closed = streams.close() println(s"Exiting application after streams close ($closed)") } catch { - case x: Exception => x.printStackTrace + case x: Exception => x.printStackTrace() } finally { println("Exiting application ..") System.exit(-1) @@ -180,13 +174,12 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro classOf[StringSerializer].getName, classOf[ByteArraySerializer].getName) userClicks - .map( - kv => - new KeyValue[String, Array[Byte]]( - kv.key, - new AvroSerde[UserClicks].serialize(UserClicks(kv.value)) + .map { kv => + new KeyValue[String, Array[Byte]]( + kv.key, + avroSerde[UserClicks].serializer().serialize("", UserClicks(kv.value)) ) - ) + } .foreach(r => sender2.writeKeyValue(userClicksTopic, r.key, r.value)) //