Upgrade to ZIO 2

build.sbt

@@ -3,15 +3,15 @@ ThisBuild / resolvers += Resolver.sonatypeRepo("snapshots")
 lazy val ProjectName = "KM8"
 lazy val ProjectOrganization = "KM8S"
 lazy val ProjectVersion = "0.2.0-SNAPSHOT"
-lazy val ProjectScalaVersion = "3.1.2"
+lazy val ProjectScalaVersion = "3.1.3"
 
 lazy val Versions = new {
 
-  val zio = "1.0.13"
-  val zioKafka = "0.17.5"
-  val zioJson = "0.2.0-M3"
-  val zioLogging = "0.5.14"
-  val zioPrelude = "1.0.0-RC8"
+  val zio = "2.0.0-RC5"
+  val zioKafka = "2.0.0-M3"
+  val zioJson = "0.3.0-RC7"
+  val zioLogging = "2.0.0-RC8"
+  val zioPrelude = "1.0.0-RC13"
 
   val kafkaProtobuf = "7.1.0"
   val javaFx = "16"

core/src/it/scala/io/km8/core/kafka/KafkaExplorerSpec.scala

@@ -1,7 +1,6 @@
 package io.km8.core.kafka
 
 import zio.*
-import zio.blocking.Blocking
 import zio.clock.Clock
 import zio.duration.*
 import zio.kafka.consumer.*
@@ -22,7 +21,6 @@ object KafkaExplorerSpec extends DefaultRunnableSpec:
 
         val layer =
           Clock.live >+>
-            Blocking.live >+>
             itlayers.kafkaContainer >+>
             itlayers.clusterConfig(clusterId) >+>
             KafkaExplorer.liveLayer >+>

core/src/it/scala/io/km8/core/kafka/KafkaProducerSpec.scala

@@ -1,7 +1,6 @@
 package io.km8.core.kafka
 
 import zio.*
-import zio.blocking.Blocking
 import zio.clock.Clock
 import zio.duration.*
 import zio.kafka.consumer.*
@@ -16,19 +15,19 @@ import org.apache.kafka.clients.consumer.OffsetResetStrategy
 
 object KafkaProducerSpec extends DefaultRunnableSpec:
 
-  private val consumerLayer: ZLayer[Has[KafkaContainer], Nothing, Has[KafkaConsumer]] =
-    Clock.live ++ Blocking.live ++ itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaConsumer.liveLayer
+  private val consumerLayer: ZLayer[KafkaContainer, Nothing, KafkaConsumer] =
+    Clock.live ++ itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaConsumer.liveLayer
 
-  private val producerLayer: ZLayer[Has[KafkaContainer], Nothing, KafkaProducer] =
-    Blocking.live ++ itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaProducer.liveLayer
+  private val producerLayer: ZLayer[KafkaContainer, Nothing, KafkaProducer] =
+    itlayers.clusterConfig(clusterId = "cluster_id") >>> KafkaProducer.liveLayer
 
-  val specLayer: ZLayer[Has[KafkaContainer], Nothing, Has[KafkaConsumer] & KafkaProducer] =
+  val specLayer: ZLayer[KafkaContainer, Nothing, KafkaConsumer & KafkaProducer] =
     consumerLayer ++ producerLayer
 
   override def spec: ZSpec[_root_.zio.test.environment.TestEnvironment, Any] =
     mainSpec
-      .provideSomeLayer[environment.TestEnvironment & Has[KafkaContainer]](specLayer ++ Clock.live)
-      .provideCustomLayerShared(Blocking.live >>> itlayers.kafkaContainer)
+      .provideSomeLayer[environment.TestEnvironment & KafkaContainer](specLayer ++ Clock.live)
+      .provideCustomLayerShared(itlayers.kafkaContainer)
 
   private val mainSpec =
     suite("Kafka services")(

core/src/it/scala/io/km8/core/kafka/itlayers.scala

@@ -3,7 +3,6 @@ package io.km8.core.kafka
 import com.dimafeng.testcontainers.KafkaContainer
 import zio.*
 import zio.clock.Clock
-import zio.blocking.*
 import zio.kafka.consumer.Consumer
 import zio.kafka.consumer.Consumer.{AutoOffsetStrategy, OffsetRetrieval}
 import zio.kafka.consumer.ConsumerSettings
@@ -12,7 +11,7 @@ import io.km8.core.config.{ClusterConfig, ClusterProperties, ClusterSettings}
 
 object itlayers:
 
-  val kafkaContainer: ZLayer[Blocking, Nothing, Has[KafkaContainer]] =
+  val kafkaContainer: ZLayer[Any, Nothing, KafkaContainer] =
     ZManaged.make {
       effectBlocking {
         val container = new KafkaContainer()
@@ -21,7 +20,7 @@ object itlayers:
       }.orDie
     }(container => effectBlocking(container.stop()).orDie).toLayer
 
-  def consumerSettings(cg: String): ZManaged[Has[KafkaContainer], Nothing, ConsumerSettings] =
+  def consumerSettings(cg: String): ZManaged[KafkaContainer, Nothing, ConsumerSettings] =
     ZIO
       .service[KafkaContainer]
       .map(c =>
@@ -31,10 +30,10 @@ object itlayers:
       )
       .toManaged_
 
-  def consumerLayer(cgroup: String): ZLayer[Has[KafkaContainer] with Clock with Blocking, Nothing, Has[Consumer]] =
+  def consumerLayer(cgroup: String): ZLayer[KafkaContainer with Clock , Nothing, Consumer] =
     consumerSettings(cgroup).flatMap(Consumer.make(_)).orDie.toLayer
 
-  def clusterConfig(clusterId: String): ZLayer[Has[KafkaContainer], Nothing, Has[ClusterConfig]] =
+  def clusterConfig(clusterId: String): ZLayer[KafkaContainer, Nothing, ClusterConfig] =
     ZIO
       .service[KafkaContainer]
       .map(kafkaContainer =>

core/src/main/scala/io/km8/core/config/ClustersConfig.scala

@@ -48,29 +48,30 @@ trait ClusterConfig {
 
 object ClustersConfig {
 
-  def readClusters = ZIO.serviceWith[ClusterConfig](_.readClusters)
-  def writeClusters(cluster: ClusterSettings) = ZIO.serviceWith[ClusterConfig](_.writeClusters(cluster))
-  def deleteCluster(clusterId: String) = ZIO.serviceWith[ClusterConfig](_.deleteCluster(clusterId))
+  def readClusters = ZIO.serviceWithZIO[ClusterConfig](_.readClusters)
+  def writeClusters(cluster: ClusterSettings) = ZIO.serviceWithZIO[ClusterConfig](_.writeClusters(cluster))
+  def deleteCluster(clusterId: String) = ZIO.serviceWithZIO[ClusterConfig](_.deleteCluster(clusterId))
 
-  lazy val liveLayer: URLayer[Has[ConfigPath] with Logging, Has[ClusterConfig]] = (ClusterConfigLive(_, _)).toLayer
+  lazy val liveLayer: URLayer[ConfigPath , ClusterConfig] =
+    ZLayer(ZIO.service[ConfigPath].map(ClusterConfigLive.apply))
 
-  case class ClusterConfigLive(configPath: ConfigPath, log: Logger[String]) extends ClusterConfig {
+  case class ClusterConfigLive(configPath: ConfigPath) extends ClusterConfig {
     private val configFilepath = configPath.path
     private def emptyProperties = ClusterProperties(List.empty)
     private def emptyPropertiesJson = emptyProperties.toJsonPretty
 
     private def writeJson(json: => String) =
-      Task(os.write.over(configFilepath, json, createFolders = true))
+      ZIO.attempt(os.write.over(configFilepath, json, createFolders = true))
 
     def readClusters: Task[ClusterProperties] =
       for {
-        b <- Task(os.exists(configFilepath))
+        b <- ZIO.attempt(os.exists(configFilepath))
         _ <- writeJson(emptyPropertiesJson).unless(b)
-        s <- Task(os.read(configFilepath))
+        s <- ZIO.attempt(os.read(configFilepath))
         r <- ZIO
                .fromEither(s.fromJson[ClusterProperties])
                .catchAll { err =>
-                 log.warn(s"Parsing error: $err") *>
+                 ZIO.logWarning(s"Parsing error: $err") *>
                    writeJson(emptyPropertiesJson).as(emptyProperties)
                }
       } yield r
@@ -85,7 +86,7 @@ object ClustersConfig {
     def deleteCluster(clusterId: String): Task[ClusterProperties] =
       for {
         c <- readClusters
-        ls <- ZIO.filterNotPar(c.clusters)(s => Task(s.id == clusterId))
+        ls <- ZIO.filterNotPar(c.clusters)(s => ZIO.attempt(s.id == clusterId))
         json = ClusterProperties(ls).toJsonPretty
         _ <- writeJson(json)
         r <- readClusters

core/src/main/scala/io/km8/core/config/ConfigPathService.scala

@@ -2,15 +2,16 @@ package io.km8.core
 package config
 
 import zio.*
-import zio.system.*
+import zio.System
+import zio.System.env
 
 lazy val EnvKey = "KAFKAMATE_ENV"
 lazy val FileName = "kafkamate.json"
 
 case class ConfigPath(path: os.Path)
 
-lazy val liveLayer: URLayer[System, Has[ConfigPath]] =
-  env(EnvKey).map {
+lazy val liveLayer: URLayer[System, ConfigPath] =
+  ZLayer.fromZIO(env(EnvKey).map {
     case Some("prod") => ConfigPath(os.root / FileName)
     case _            => ConfigPath(os.pwd / FileName)
-  }.orDie.toLayer
+  }.orDie)

core/src/main/scala/io/km8/core/kafka/KafkaConsumer.scala

@@ -2,9 +2,9 @@ package io.km8.core
 package kafka
 
 import zio.*
-import zio.blocking.Blocking
-import zio.clock.Clock
-import zio.duration.*
+
+import zio.Clock
+
 import zio.kafka.consumer.*
 import zio.kafka.consumer.Consumer.*
 import zio.kafka.serde.Deserializer
@@ -28,19 +28,23 @@ trait KafkaConsumer {
 
 object KafkaConsumer {
 
-  lazy val liveLayer: URLayer[Clock with Blocking with Has[ClusterConfig], Has[KafkaConsumer]] =
-    (KafkaConsumerLive(_, _, _)).toLayer
+  lazy val liveLayer: URLayer[Clock with ClusterConfig, KafkaConsumer] =
+    ZLayer {
+      for {
+        clock <- ZIO.service[Clock]
+        config <- ZIO.service[ClusterConfig]
+      } yield KafkaConsumerLive(clock, config)
+    }
 
-  def consume(request: ConsumeRequest): ZStream[Has[KafkaConsumer], Throwable, Message] =
-    ZStream.accessStream[Has[KafkaConsumer]](_.get.consume(request))
+  def consume(request: ConsumeRequest): ZStream[KafkaConsumer, Throwable, Message] =
+    ZStream.environmentWithStream[KafkaConsumer](_.get.consume(request))
 
   case class KafkaConsumerLive(
-    clock: Clock.Service,
-    blocking: Blocking.Service,
+    clock: Clock,
     clustersConfigService: ClusterConfig)
       extends KafkaConsumer {
+
     private val clockLayer = ZLayer.succeed(clock)
-    private val blockingLayer = ZLayer.succeed(blocking)
 
     // TODO Ciprian transform to enum or reuse the Consumer enum + serdes
     private def extractOffsetStrategy(offsetValue: String): AutoOffsetStrategy =
@@ -55,7 +59,7 @@ object KafkaConsumer {
         .map(_.asTry)
 
     private def consumerSettings(config: ClusterSettings, offsetStrategy: String): Task[ConsumerSettings] =
-      Task {
+      ZIO.attempt {
         val uuid = UUID.randomUUID().toString
         ConsumerSettings(config.kafkaHosts)
           .withGroupId(s"group-kafkamate-$uuid")
@@ -68,19 +72,19 @@ object KafkaConsumer {
     private def makeConsumerLayer(
       clusterId: String,
       offsetStrategy: String
-    ): ZLayer[Clock & Blocking, Throwable, Has[Consumer]] =
-      ZLayer.fromManaged {
+    ): ZLayer[Any, Throwable, Consumer] =
+      ZLayer.scoped {
         for {
-          cs <- clustersConfigService.getCluster(clusterId).toManaged_
-          settings <- consumerSettings(cs, offsetStrategy).toManaged_
+          cs <- clustersConfigService.getCluster(clusterId)
+          settings <- consumerSettings(cs, offsetStrategy)
           consumer <- Consumer.make(settings)
         } yield consumer
       }
 
     def consume(request: ConsumeRequest): ZStream[Any, Throwable, Message] = {
       def consumer[T](
         valueDeserializer: Deserializer[Any, Try[T]]
-      ): ZStream[Has[Consumer], Throwable, Message] = Consumer
+      ): ZStream[Consumer, Throwable, Message] = Consumer
         .subscribeAnd(Subscription.topics(request.topicName))
         .plainStream(Deserializer.string, valueDeserializer)
         .collect {
@@ -98,7 +102,7 @@ object KafkaConsumer {
                 .orElseFail(new Exception("SchemaRegistry url was not provided!"))
             )
           ZStream
-            .fromEffect(protoSettings.flatMap(protobufDeserializer))
+            .fromZIO(protoSettings.flatMap(protobufDeserializer))
             .flatMap(consumer)
         case _ => consumer(Deserializer.string.asTry)
       }
@@ -114,7 +118,7 @@ object KafkaConsumer {
         else withFilter.take(request.maxResults)
 
       withFilterLimit.provideLayer(
-        clockLayer ++ blockingLayer >>>
+        clockLayer >>>
           makeConsumerLayer(request.clusterId, request.offsetStrategy)
       )
     }