Non-effectful serialization and deserialization

zio-kafka-test/src/test/scala/zio/kafka/AdminSpec.scala

@@ -219,7 +219,7 @@ object AdminSpec extends ZIOSpecDefaultSlf4j with KafkaRandom {
 
           def consumeAndCommit(count: Long) =
             Consumer
-              .partitionedStream[Kafka, String, String](Subscription.Topics(Set(topic)), Serde.string, Serde.string)
+              .partitionedStream[Any, String, String](Subscription.Topics(Set(topic)), Serde.string, Serde.string)
               .flatMapPar(partitionCount)(_._2)
               .take(count)
               .transduce(ZSink.collectAllN[CommittableRecord[String, String]](20))
@@ -293,7 +293,7 @@ object AdminSpec extends ZIOSpecDefaultSlf4j with KafkaRandom {
 
         def consumeAndCommit(count: Long, topic: String, groupId: String) =
           Consumer
-            .plainStream[Kafka, String, String](Subscription.Topics(Set(topic)), Serde.string, Serde.string)
+            .plainStream[Any, String, String](Subscription.Topics(Set(topic)), Serde.string, Serde.string)
             .take(count)
             .foreach(_.offset.commit)
             .provideSomeLayer[Kafka](consumer(topic, Some(groupId)))
@@ -336,7 +336,7 @@ object AdminSpec extends ZIOSpecDefaultSlf4j with KafkaRandom {
 
         def consumeAndCommit(count: Long, topic: String, groupId: String) =
           Consumer
-            .plainStream[Kafka, String, String](Subscription.Topics(Set(topic)), Serde.string, Serde.string)
+            .plainStream[Any, String, String](Subscription.Topics(Set(topic)), Serde.string, Serde.string)
             .take(count)
             .foreach(_.offset.commit)
             .provideSomeLayer[Kafka](consumer(topic, Some(groupId)))

zio-kafka-test/src/test/scala/zio/kafka/Benchmarks.scala

@@ -25,7 +25,7 @@ object PopulateTopic extends ZIOAppDefault {
   override def run: ZIO[Any, Throwable, Unit] =
     dataStream(872000).map { case (k, v) =>
       new ProducerRecord("inputs-topic", null, null, k, v)
-    }.mapChunksZIO(Producer.produceChunkAsync[Any, String, String](_, Serde.string, Serde.string).map(Chunk(_)))
+    }.mapChunksZIO(Producer.produceChunkAsync[String, String](_, Serde.string, Serde.string).map(Chunk(_)))
       .mapZIOPar(5)(_.flatMap(chunk => Console.printLine(s"Wrote chunk of ${chunk.size}")))
       .runDrain
       .provide(

zio-kafka-test/src/test/scala/zio/kafka/serde/DeserializerSpec.scala

@@ -2,13 +2,12 @@ package zio.kafka.serde
 
 import org.apache.kafka.common.header.internals.RecordHeaders
 import zio._
+import zio.kafka.ZIOSpecDefaultSlf4j
 import zio.test.Assertion._
 import zio.test._
-import zio.ZAny
-import zio.kafka.ZIOSpecDefaultSlf4j
 
 object DeserializerSpec extends ZIOSpecDefaultSlf4j {
-  override def spec: Spec[ZAny with Any, Throwable] = suite("Deserializer")(
+  override def spec: Spec[TestEnvironment with Scope, Any] = suite("Deserializer")(
     suite("asOption")(
       test("deserialize to None when value is null") {
         assertZIO(stringDeserializer.asOption.deserialize("topic1", new RecordHeaders, null))(isNone)

zio-kafka-test/src/test/scala/zio/kafka/serde/SerdeSpec.scala

@@ -1,6 +1,7 @@
 package zio.kafka.serde
 
 import org.apache.kafka.common.header.internals.RecordHeaders
+import zio.Scope
 import zio.kafka.ZIOSpecDefaultSlf4j
 import zio.test.Assertion._
 import zio.test._
@@ -15,7 +16,7 @@ object SerdeSpec extends ZIOSpecDefaultSlf4j {
 
   private val anyBytes = Gen.listOf(Gen.byte).map(bytes => new org.apache.kafka.common.utils.Bytes(bytes.toArray))
 
-  override def spec: Spec[Any, Throwable] = suite("Serde")(
+  override def spec: Spec[TestEnvironment with Scope, Any] = suite("Serde")(
     testSerde(Serde.string, Gen.string),
     testSerde(Serde.int, Gen.int),
     testSerde(Serde.short, Gen.short),
@@ -27,20 +28,20 @@ object SerdeSpec extends ZIOSpecDefaultSlf4j {
     testSerde(Serde.byteArray, Gen.listOf(Gen.byte).map(_.toArray)),
     suite("asOption")(
       test("serialize and deserialize None values to null and visa versa") {
-        val serde = testDataStructureSerde.asOption
+        val serde      = testDataStructureSerde.asOption
+        val serialized = serde.serialize("topic1", new RecordHeaders, None)
         for {
-          serialized   <- serde.serialize("topic1", new RecordHeaders, None)
           deserialized <- serde.deserialize("topic1", new RecordHeaders, serialized)
         } yield assert(serialized)(isNull) && assert(deserialized)(isNone)
       }
     )
   )
 
-  private def testSerde[R, A](serde: Serde[Any, A], gen: Gen[R, A])(implicit clsTag: ClassTag[A]) =
+  private def testSerde[R, A](serde: Serde[Throwable, A], gen: Gen[R, A])(implicit clsTag: ClassTag[A]) =
     test(s"serialize and deserialize ${clsTag.runtimeClass.getSimpleName}") {
       check(gen) { value =>
+        val serialized = serde.serialize("topic1", new RecordHeaders, value)
         for {
-          serialized   <- serde.serialize("topic1", new RecordHeaders, value)
           deserialized <- serde.deserialize("topic1", new RecordHeaders, serialized)
         } yield assert(deserialized)(equalTo(value))
       }

zio-kafka-test/src/test/scala/zio/kafka/serde/SerializerSpec.scala

@@ -3,25 +3,27 @@ package zio.kafka.serde
 import org.apache.kafka.common.header.internals.RecordHeaders
 import zio.test.Assertion._
 import zio.test._
-import zio.ZAny
+import zio.{ ZAny, ZIO }
 import zio.kafka.ZIOSpecDefaultSlf4j
 
 object SerializerSpec extends ZIOSpecDefaultSlf4j {
   override def spec: Spec[ZAny with Any, Throwable] = suite("Serializer")(
     suite("asOption")(
       test("serialize None values to null") {
-        assertZIO(stringSerializer.asOption.serialize("topic1", new RecordHeaders, None))(isNull)
+        assert(stringSerializer.asOption.serialize("topic1", new RecordHeaders, None))(isNull)
       },
       test("serialize Some values") {
         check(Gen.string) { string =>
           assertZIO(
-            stringSerializer.asOption.serialize("topic1", new RecordHeaders, Some(string)).map(new String(_, "UTF-8"))
+            ZIO
+              .succeed(stringSerializer.asOption.serialize("topic1", new RecordHeaders, Some(string)))
+              .map(new String(_, "UTF-8"))
           )(
             equalTo(string)
           )
         }
       }
     )
   )
-  private lazy val stringSerializer: Serializer[Any, String] = Serde.string
+  private lazy val stringSerializer: Serializer[String] = Serde.string
 }

zio-kafka-testkit/src/main/scala/zio/kafka/testkit/KafkaTestUtils.scala

@@ -70,7 +70,7 @@ object KafkaTestUtils {
     key: String,
     message: String
   ): ZIO[Producer, Throwable, RecordMetadata] =
-    Producer.produce[Any, String, String](new ProducerRecord(topic, key, message), Serde.string, Serde.string)
+    Producer.produce[String, String](new ProducerRecord(topic, key, message), Serde.string, Serde.string)
 
   /**
    * Utility function to produce many messages in give Partition of a Topic.
@@ -81,7 +81,7 @@ object KafkaTestUtils {
     kvs: Iterable[(String, String)]
   ): ZIO[Producer, Throwable, Chunk[RecordMetadata]] =
     Producer
-      .produceChunk[Any, String, String](
+      .produceChunk[String, String](
         Chunk.fromIterable(kvs.map { case (k, v) =>
           new ProducerRecord(topic, partition, null, k, v)
         }),
@@ -97,7 +97,7 @@ object KafkaTestUtils {
     kvs: Iterable[(String, String)]
   ): ZIO[Producer, Throwable, Chunk[RecordMetadata]] =
     Producer
-      .produceChunk[Any, String, String](
+      .produceChunk[String, String](
         Chunk.fromIterable(kvs.map { case (k, v) =>
           new ProducerRecord(topic, k, v)
         }),

zio-kafka/src/main/scala/zio/kafka/consumer/CommittableRecord.scala

@@ -3,17 +3,17 @@ package zio.kafka.consumer
 import org.apache.kafka.clients.consumer.{ ConsumerGroupMetadata, ConsumerRecord, OffsetAndMetadata }
 import org.apache.kafka.common.TopicPartition
 import zio.kafka.serde.Deserializer
-import zio.{ RIO, Task }
+import zio.{ IO, Task }
 
 final case class CommittableRecord[K, V](
   record: ConsumerRecord[K, V],
   private val commitHandle: Map[TopicPartition, OffsetAndMetadata] => Task[Unit],
   private val consumerGroupMetadata: Option[ConsumerGroupMetadata]
 ) {
-  def deserializeWith[R, K1, V1](
-    keyDeserializer: Deserializer[R, K1],
-    valueDeserializer: Deserializer[R, V1]
-  )(implicit ev1: K <:< Array[Byte], ev2: V <:< Array[Byte]): RIO[R, CommittableRecord[K1, V1]] =
+  def deserializeWith[E, K1, V1](
+    keyDeserializer: Deserializer[E, K1],
+    valueDeserializer: Deserializer[E, V1]
+  )(implicit ev1: K <:< Array[Byte], ev2: V <:< Array[Byte]): IO[E, CommittableRecord[K1, V1]] =
     for {
       key   <- keyDeserializer.deserialize(record.topic(), record.headers(), record.key())
       value <- valueDeserializer.deserialize(record.topic(), record.headers(), record.value())