Cross compile for Scala 3

jmast · jmast · commit 7a5a0050d99e · 2025-03-11T14:31:29.000+01:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -10,7 +10,8 @@ jobs:
     strategy:
       matrix:
         scala:
-          - 2.13.14
+          - 2.13.16
+          - 3.3.5
 
     steps:
     - uses: actions/checkout@v4
diff --git a/build.sbt b/build.sbt
@@ -4,14 +4,17 @@ ThisBuild / versionScheme := Some("early-semver")
 ThisBuild / evictionErrorLevel := Level.Warn
 ThisBuild / versionPolicyIntention := Compatibility.BinaryCompatible
 
+lazy val Scala3Version = "3.3.5"
+lazy val Scala2Version = "2.13.16"
+
 lazy val commonSettings = Seq(
   organization := "com.evolutiongaming",
-  homepage := Some(new URL("https://github.com/evolution-gaming/kafka-flow")),
+  homepage := Some(url("https://github.com/evolution-gaming/kafka-flow")),
   startYear := Some(2019),
   organizationName := "Evolution Gaming",
   organizationHomepage := Some(url("https://evolution.com/")),
   publishTo := Some(Resolver.evolutionReleases),
-  scalaVersion := "2.13.16",
+  scalaVersion := Scala2Version,
   licenses := Seq(("MIT", url("https://opensource.org/licenses/MIT"))),
   testFrameworks += new TestFramework("munit.Framework"),
   testOptions += Tests.Argument(new TestFramework("munit.Framework"), "+l"),
@@ -20,10 +23,26 @@ lazy val commonSettings = Seq(
   libraryDependencySchemes ++= Seq(
     "org.scala-lang.modules" %% "scala-java8-compat" % "always"
   ),
-  addCompilerPlugin("org.typelevel" %% "kind-projector" % "0.13.3" cross CrossVersion.full),
-  scalacOptions ++= Seq("-Xsource:3"),
+  libraryDependencies ++= crossSettings(
+    scalaVersion.value,
+    if3 = Nil,
+    if2 = List(compilerPlugin("org.typelevel" %% "kind-projector" % "0.13.3" cross CrossVersion.full)),
+  ),
+  scalacOptions ++= crossSettings(
+    scalaVersion.value,
+    if3 = List("-Ykind-projector", "-language:implicitConversions", "-explain", "-deprecation"),
+    if2 = List("-Xsource:3")
+  ),
+  crossScalaVersions := Seq(Scala2Version, Scala3Version)
 )
 
+def crossSettings[T](scalaVersion: String, if3: List[T], if2: List[T]) =
+  CrossVersion.partialVersion(scalaVersion) match {
+    case Some((3, _))       => if3
+    case Some((2, 12 | 13)) => if2
+    case _                  => Nil
+  }
+
 lazy val root = (project in file("."))
   .aggregate(
     core,
@@ -38,7 +57,8 @@ lazy val root = (project in file("."))
   .settings(commonSettings)
   .settings(
     name := "kafka-flow",
-    publish / skip := true
+    publish / skip := true,
+    crossScalaVersions := Nil,
   )
 
 lazy val core = (project in file("core"))
@@ -58,10 +78,14 @@ lazy val core = (project in file("core"))
       sstream,
       random,
       retry,
-      Scodec.core,
       Scodec.bits,
       Testing.munit % Test,
     ),
+    libraryDependencies ++= crossSettings(
+      scalaVersion.value,
+      if3 = List(Scodec.coreScala3),
+      if2 = List(Scodec.coreScala213)
+    )
   )
 
 lazy val `core-it-tests` = (project in file("core-it-tests"))
@@ -98,6 +122,11 @@ lazy val `persistence-cassandra` = (project in file("persistence-cassandra"))
       scassandra,
       cassandraSync,
     ),
+    libraryDependencies ++= crossSettings(
+      scalaVersion.value,
+      if3 = List(PureConfig.GenericScala3),
+      if2 = Nil,
+    )
   )
 
 lazy val `persistence-cassandra-it-tests` = (project in file("persistence-cassandra-it-tests"))
@@ -146,7 +175,8 @@ lazy val journal = (project in file("kafka-journal"))
       KafkaJournal.journal,
       KafkaJournal.persistence,
       Testing.munit % Test,
-    )
+    ),
+    crossScalaVersions := Seq(Scala2Version),
   )
 
 lazy val docs = (project in file("kafka-flow-docs"))
diff --git a/core-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/ForAllKafkaSuite.scala b/core-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/ForAllKafkaSuite.scala
@@ -26,7 +26,7 @@ abstract class ForAllKafkaSuite extends FunSuite with TestContainersFixtures {
     override def beforeAll(): Unit = {
       val config =
         ConsumerConfig(common = CommonConfig(bootstrapServers = NonEmptyList.one(kafka.container.bootstrapServers)))
-      implicit val logOf = LogOf.slf4j[IO].unsafeRunSync()
+      implicit val logOf: LogOf[IO] = LogOf.slf4j[IO].unsafeRunSync()
       val result = KafkaModule.of[IO]("KafkaSuite", config, CollectorRegistry.empty[IO]).allocated.unsafeRunSync()
       moduleRef.set(result)
     }
diff --git a/core/src/main/scala-3/com/evolutiongaming/kafka/flow/persistence/compression/CodecOps.scala b/core/src/main/scala-3/com/evolutiongaming/kafka/flow/persistence/compression/CodecOps.scala
@@ -0,0 +1,15 @@
+package com.evolutiongaming.kafka.flow.persistence.compression
+
+import scodec.Codec
+import scodec.Codec.inlineImplementations
+
+extension[A, B](codecA: Codec[A])
+
+/** Combines this Codec with another one, added for compatibility with Scala 2.13 version of scodec-core.
+  * @param codecB
+  *   Codec for B
+  * @return
+  *   Codec for tuple of A and B
+  */
+def ~(codecB: Codec[B]): Codec[(A, B)] =
+  codecA :: codecB
diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/PartitionFlow.scala
@@ -234,7 +234,7 @@ object PartitionFlow {
         case (_, Right(value)) => value.context.holding
         case (key, Left(_)) =>
           log.error(s"trying to compute offset to commit but value for key $key is not ready").as(none[Offset])
-      }(pickMinOffset))
+      })
     }
 
     def offsetToCommit(getMinOffset: F[Option[Offset]]): F[Option[Offset]] = for {
@@ -338,13 +338,14 @@ object PartitionFlow {
 
   }
 
-  private[this] val pickMinOffset = CommutativeMonoid.instance[Option[Offset]](
-    none[Offset],
-    (x, y) =>
-      (x, y) match {
-        case (Some(x), Some(y)) => Some(x min y)
-        case (x @ Some(_), _)   => x
-        case (_, y)             => y
-      }
-  )
+  private[this] implicit val pickMinOffset: CommutativeMonoid[Option[Offset]] =
+    CommutativeMonoid.instance[Option[Offset]](
+      none[Offset],
+      (x, y) =>
+        (x, y) match {
+          case (Some(x), Some(y)) => Some(x min y)
+          case (x @ Some(_), _)   => x
+          case (_, y)             => y
+        }
+    )
 }
diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/TopicFlow.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/TopicFlow.scala
@@ -15,7 +15,6 @@ import com.evolutiongaming.skafka.consumer.ConsumerRecords
 import scodec.bits.ByteVector
 
 import scala.collection.immutable.SortedSet
-import scala.annotation.nowarn
 
 trait TopicFlow[F[_]] {
 
@@ -82,7 +81,6 @@ object TopicFlow {
 
     val acquire = new TopicFlow[F] {
 
-      @nowarn("cat=unused") // for some reason compiler thinks that `flow` and `partitionRecords` are unused
       def apply(records: ConsumerRecords[String, ByteVector]) = {
 
         for {
diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/kafka/KafkaModule.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/kafka/KafkaModule.scala
@@ -35,17 +35,17 @@ object KafkaModule {
     config: ConsumerConfig,
     registry: CollectorRegistry[F]
   ): Resource[F, KafkaModule[F]] = {
-    implicit val measureDuration = MeasureDuration.fromClock[F](Clock[F])
+    implicit val measureDuration: MeasureDuration[F] = MeasureDuration.fromClock[F](Clock[F])
     for {
       producerMetrics <- ProducerMetrics.of(registry)
       consumerMetrics <- ConsumerMetrics.of(registry)
       _producerOf      = RawProducerOf.apply1[F](producerMetrics(applicationId).some)
       _consumerOf      = RawConsumerOf.apply1[F](consumerMetrics(applicationId).some)
 
       _healthCheck <- {
-        implicit val randomIdOf = RandomIdOf.uuid[F]
-        implicit val consumerOf = _consumerOf
-        implicit val producerOf = _producerOf
+        implicit val randomIdOf: RandomIdOf[F] = RandomIdOf.uuid[F]
+        implicit val consumerOf: RawConsumerOf[F] = _consumerOf
+        implicit val producerOf: RawProducerOf[F] = _producerOf
 
         val commonConfig = config.common.copy(clientId = config.common.clientId.map(id => s"$id-HealthCheck"))
 
diff --git a/core/src/main/scala/com/evolutiongaming/kafka/flow/persistence/compression/HeaderAndPayload.scala b/core/src/main/scala/com/evolutiongaming/kafka/flow/persistence/compression/HeaderAndPayload.scala
@@ -11,7 +11,7 @@ private[compression] object HeaderAndPayload {
   private val codec = variableSizeBytes(int32, bytes) ~ bytes
 
   def toBytes[F[_]: MonadThrow](header: ByteVector, payload: ByteVector): F[ByteVector] = {
-    codec.encode(header ~ payload) match {
+    codec.encode((header, payload)) match {
       case Attempt.Successful(value) => value.bytes.pure[F]
       case Attempt.Failure(e)        => CompressionError(e.message).raiseError
     }
diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/ConsumerFlowSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/ConsumerFlowSpec.scala
@@ -15,7 +15,7 @@ import munit.FunSuite
 import scodec.bits.ByteVector
 
 import scala.concurrent.duration.*
-import scala.util.{Success, Try}
+import scala.util.Try
 
 class ConsumerFlowSpec extends FunSuite {
 
@@ -29,7 +29,7 @@ class ConsumerFlowSpec extends FunSuite {
       Command.Records()
     )
 
-    val Success(result) = ConstFixture.app(topic).runS(Context(commands = commands))
+    val result = ConstFixture.app(topic).runS(Context(commands = commands)).get
 
     assertEquals(
       result.actions.reverse,
@@ -64,7 +64,7 @@ class ConsumerFlowSpec extends FunSuite {
       Command.Records()
     )
 
-    val Success(result) = ConstFixture.app(topic1, topic2).runS(Context(commands = commands))
+    val result = ConstFixture.app(topic1, topic2).runS(Context(commands = commands)).get
 
     assertEquals(
       result.actions.reverse,
diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/KeyFlowSpec.scala
@@ -84,7 +84,7 @@ class KeyFlowSpec extends FunSuite {
       Offset.unsafe(1) -> 7
     }
     val timerFlowOf = TimerFlowOf.unloadOrphaned[SyncIO]()
-    implicit val context = new KeyContext[SyncIO] {
+    implicit val context: KeyContext[SyncIO] = new KeyContext[SyncIO] {
       def holding              = none[Offset].pure[SyncIO]
       def hold(offset: Offset) = SyncIO.unit
       def remove               = removeCalled.set(true)
@@ -134,7 +134,7 @@ class KeyFlowSpec extends FunSuite {
     val fold        = FoldOption.empty[SyncIO, State, ConsumerRecord[String, ByteVector]]
     val timerFlowOf = TimerFlowOf.unloadOrphaned[SyncIO]()
 
-    implicit val context = new KeyContext[SyncIO] {
+    implicit val context: KeyContext[SyncIO] = new KeyContext[SyncIO] {
       def holding              = none[Offset].pure[SyncIO]
       def hold(offset: Offset) = SyncIO.unit
       def remove               = removeCalled.set(true)
diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/PartitionFlowSpec.scala
@@ -394,7 +394,7 @@ class PartitionFlowSpec extends FunSuite {
 
   def setupRemapKeyTest(remapKey: RemapKey[IO], initialData: Map[KafkaKey, String]) = {
     import com.evolutiongaming.kafka.flow.effect.CatsEffectMtlInstances.*
-    implicit val logOf = LogOf.empty[IO]
+    implicit val logOf: LogOf[IO] = LogOf.empty[IO]
     logOf.apply(classOf[PartitionFlowSpec]).toResource.flatMap { implicit log =>
       val committedOffset  = Ref.unsafe[IO, Offset](Offset.min)
       val keyStorage       = Ref.unsafe[IO, Set[KafkaKey]](initialData.keySet)
diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/registry/EntityRegistryTest.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/registry/EntityRegistryTest.scala
@@ -14,9 +14,7 @@ import munit.FunSuite
 import scodec.bits.ByteVector
 
 import java.nio.charset.StandardCharsets
-import scala.annotation.nowarn
 
-@nowarn("cat=unused") // for some reason compiler thinks that values in the test are unused
 class EntityRegistryTest extends FunSuite {
   implicit val logOf: LogOf[IO] = LogOf.empty[IO]
   implicit val log: Log[IO]     = Log.empty[IO]
diff --git a/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala b/core/src/test/scala/com/evolutiongaming/kafka/flow/timer/TimerFlowOfSpec.scala
@@ -572,7 +572,7 @@ object TimerFlowSpec {
   )
 
   object Context {
-    val lens: GenLens[Context] = GenLens[Context]
+    val lens = GenLens[Context]
   }
 
   class ConstFixture {
diff --git a/metrics/src/main/scala/com/evolutiongaming/kafka/flow/FlowMetrics.scala b/metrics/src/main/scala/com/evolutiongaming/kafka/flow/FlowMetrics.scala
@@ -19,7 +19,6 @@ import com.evolutiongaming.kafka.flow.snapshot.SnapshotDatabaseMetrics.*
 import com.evolutiongaming.skafka.consumer.ConsumerRecord
 import com.evolutiongaming.smetrics.CollectorRegistry
 import scodec.bits.ByteVector
-import scala.annotation.nowarn
 
 trait FlowMetrics[F[_]] {
 
@@ -40,7 +39,6 @@ object FlowMetrics {
 
   def apply[F[_]](implicit F: FlowMetrics[F]): FlowMetrics[F] = F
 
-  @nowarn("cat=unused") // for some reason compiler thinks that `persistentModule` is unused
   def of[F[_]: Monad: MeasureDuration](registry: CollectorRegistry[F]): Resource[F, FlowMetrics[F]] = for {
     keyDatabase      <- keyDatabaseMetricsOf[F].apply(registry)
     journalDatabase  <- journalDatabaseMetricsOf[F].apply(registry)
diff --git a/metrics/src/test/scala/com/evolutiongaming/kafka/flow/FoldMetricsSpec.scala b/metrics/src/test/scala/com/evolutiongaming/kafka/flow/FoldMetricsSpec.scala
@@ -14,7 +14,7 @@ class FoldMetricsSpec extends FunSuite {
   type F[T] = Option[T]
 
   test("having MetricsKOf enables withCollectorRegistry syntax") {
-    implicit val measureDuration     = MeasureDuration.empty[F]
+    implicit val measureDuration: MeasureDuration[F] = MeasureDuration.empty[F]
     val fold: FoldOptionCons[F, Int] = FoldOption.empty[F, Int, ConsumerRecord[String, ByteVector]]
     fold.withCollectorRegistry(CollectorRegistry.empty[F])
   }
diff --git a/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/CassandraSpec.scala b/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/CassandraSpec.scala
@@ -22,7 +22,7 @@ abstract class CassandraSpec extends FunSuite {
     override def apply(): CassandraModule[IO] = moduleRef.get()._1
 
     override def beforeAll(): Unit = {
-      implicit val logOf = LogOf.slf4j[IO].unsafeRunSync()
+      implicit val logOf: LogOf[IO] = LogOf.slf4j[IO].unsafeRunSync()
 
       val container = CassandraContainerResource.cassandra.cassandraContainer
       val result: (CassandraModule[IO], IO[Unit]) =
diff --git a/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/cassandra/CassandraPersistence.scala b/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/cassandra/CassandraPersistence.scala
@@ -9,6 +9,7 @@ import com.evolutiongaming.kafka.flow.journal.CassandraJournals
 import com.evolutiongaming.kafka.flow.key.{CassandraKeys, KeySegments}
 import com.evolutiongaming.kafka.flow.persistence.PersistenceModule
 import com.evolutiongaming.kafka.flow.snapshot.CassandraSnapshots
+import com.evolutiongaming.skafka.{FromBytes, ToBytes}
 import com.evolutiongaming.{scassandra, skafka}
 
 import scala.util.Try
@@ -45,8 +46,8 @@ object CassandraPersistence {
     keysSegments: KeySegments
   )(implicit fromBytes: skafka.FromBytes[Try, S], toBytes: skafka.ToBytes[Try, S]): F[PersistenceModule[F, S]] = {
     val fromTry             = FunctionK.liftFunction[Try, F](MonadThrow[F].fromTry)
-    implicit val _fromBytes = fromBytes mapK fromTry
-    implicit val _toBytes   = toBytes mapK fromTry
+    implicit val _fromBytes: FromBytes[F, S] = fromBytes mapK fromTry
+    implicit val _toBytes: ToBytes[F, S] = toBytes mapK fromTry
     withSchemaF(session, sync, consistencyOverrides, keysSegments)
   }
 
diff --git a/persistence-kafka-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/ForAllKafkaSuite.scala b/persistence-kafka-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/ForAllKafkaSuite.scala
@@ -26,7 +26,7 @@ abstract class ForAllKafkaSuite extends FunSuite with TestContainersFixtures {
     override def beforeAll(): Unit = {
       val config =
         ConsumerConfig(common = CommonConfig(bootstrapServers = NonEmptyList.one(kafka.container.bootstrapServers)))
-      implicit val logOf = LogOf.slf4j[IO].unsafeRunSync()
+      implicit val logOf: LogOf[IO] = LogOf.slf4j[IO].unsafeRunSync()
       val result = KafkaModule.of[IO]("KafkaSuite", config, CollectorRegistry.empty[IO]).allocated.unsafeRunSync()
       moduleRef.set(result)
     }
diff --git a/persistence-kafka/src/main/scala/com/evolutiongaming/kafka/flow/kafkapersistence/KafkaPartitionPersistence.scala b/persistence-kafka/src/main/scala/com/evolutiongaming/kafka/flow/kafkapersistence/KafkaPartitionPersistence.scala
@@ -57,12 +57,12 @@ object KafkaPartitionPersistence {
     case _ => map // ignore records with no key for now
   }
 
-  private[kafkapersistence] def readSnapshots[F[_]: BracketThrowable: FromBytes[*[_], String]: Log](
+  private[kafkapersistence] def readSnapshots[F[_]: BracketThrowable: Log](
     consumerOf: ConsumerOf[F],
     consumerConfig: ConsumerConfig,
     snapshotTopic: Topic,
     partition: Partition
-  ): F[BytesByKey] = {
+  )(implicit fromBytes: FromBytes[F, String]): F[BytesByKey] = {
     consumerOf
       .apply[String, ByteVector](
         consumerConfig.copy(
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
@@ -9,9 +9,9 @@ object Dependencies {
   val skafka            = "com.evolutiongaming" %% "skafka"              % "17.1.3"
   val sstream           = "com.evolutiongaming" %% "sstream"             % "1.0.2"
   val scassandra        = "com.evolutiongaming" %% "scassandra"          % "5.3.0"
-  val cassandraSync     = "com.evolutiongaming" %% "cassandra-sync"      % "3.0.0"
-  val random            = "com.evolutiongaming" %% "random"              % "1.0.0"
-  val retry             = "com.evolutiongaming" %% "retry"               % "3.0.1"
+  val cassandraSync     = "com.evolutiongaming" %% "cassandra-sync"      % "3.1.0"
+  val random            = "com.evolution"       %% "random"              % "1.0.5"
+  val retry             = "com.evolutiongaming" %% "retry"               % "3.1.0"
   val playJsonJsoniter  = "com.evolution"       %% "play-json-jsoniter"  % "1.1.1"
 
   object Cats {
@@ -24,8 +24,9 @@ object Dependencies {
   }
 
   object Scodec {
-    val core = "org.scodec" %% "scodec-core" % "1.11.7"
-    val bits = "org.scodec" %% "scodec-bits" % "1.1.20"
+    val coreScala213 = "org.scodec" %% "scodec-core" % "1.11.10"
+    val coreScala3   = "org.scodec" %% "scodec-core" % "2.3.2"
+    val bits         = "org.scodec" %% "scodec-bits" % "1.2.1"
   }
 
   object KafkaJournal {
@@ -40,6 +41,11 @@ object Dependencies {
     val `macro`         = "dev.optics" %% "monocle-macro" % version
   }
 
+  object PureConfig {
+    private val version = "0.17.8"
+    lazy val GenericScala3 = "com.github.pureconfig" %% "pureconfig-generic-scala3" % version
+  }
+
   object Testing {
     val munit = "org.scalameta" %% "munit" % "1.1.0"
 

Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,7 @@ abstract class ForAllKafkaSuite extends FunSuite with TestContainersFixtures {`
`26`	`26`	`override def beforeAll(): Unit = {`
`27`	`27`	`val config =`
`28`	`28`	`ConsumerConfig(common = CommonConfig(bootstrapServers = NonEmptyList.one(kafka.container.bootstrapServers)))`
`29`		`- implicit val logOf = LogOf.slf4j[IO].unsafeRunSync()`
	`29`	`+ implicit val logOf: LogOf[IO] = LogOf.slf4j[IO].unsafeRunSync()`
`30`	`30`	`val result = KafkaModule.of[IO]("KafkaSuite", config, CollectorRegistry.empty[IO]).allocated.unsafeRunSync()`
`31`	`31`	`moduleRef.set(result)`
`32`	`32`	`}`
Original file line number	Diff line number	Diff line change
`@@ -11,7 +11,7 @@ private[compression] object HeaderAndPayload {`
`11`	`11`	`private val codec = variableSizeBytes(int32, bytes) ~ bytes`
`12`	`12`
`13`	`13`	`def toBytes[F[_]: MonadThrow](header: ByteVector, payload: ByteVector): F[ByteVector] = {`
`14`		`- codec.encode(header ~ payload) match {`
	`14`	`+ codec.encode((header, payload)) match {`
`15`	`15`	`case Attempt.Successful(value) => value.bytes.pure[F]`
`16`	`16`	`case Attempt.Failure(e) => CompressionError(e.message).raiseError`
`17`	`17`	`}`