Cross compile for Scala 3

Author: Z1kkurat

.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
@@ -32,13 +33,13 @@ jobs:
 
     - name: Run tests ${{ matrix.scala }}
       if: success() 
-      run: sbt ++${{ matrix.scala }} clean coverage test docs/mdoc versionPolicyCheck
+      run: sbt clean coverage "++${{ matrix.scala }} test" docs/mdoc "++${{ matrix.scala }} versionPolicyCheck"
 
     - name: Report test coverage
       if: success() && github.repository == 'evolution-gaming/kafka-flow'
       env:
         COVERALLS_REPO_TOKEN: ${{ secrets.COVERALLS_REPO_TOKEN }}
-      run: sbt ++${{ matrix.scala }} coverageReport coverageAggregate coveralls
+      run: sbt "++${{ matrix.scala }} coverageReport" coverageAggregate coveralls
 
     - name: Publish documentation / Setup Node
       if: success() 

build.sbt

@@ -4,14 +4,18 @@ 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",
+  crossScalaVersions := Seq(Scala2Version, Scala3Version),
+  scalaVersion := crossScalaVersions.value.head,
   licenses := Seq(("MIT", url("https://opensource.org/licenses/MIT"))),
   testFrameworks += new TestFramework("munit.Framework"),
   testOptions += Tests.Argument(new TestFramework("munit.Framework"), "+l"),
@@ -20,10 +24,25 @@ 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")
+  ),
 )
 
+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,9 @@ lazy val root = (project in file("."))
   .settings(commonSettings)
   .settings(
     name := "kafka-flow",
-    publish / skip := true
+    publish / skip := true,
+    crossScalaVersions := Nil,
+    scalaVersion := Scala2Version,
   )
 
 lazy val core = (project in file("core"))
@@ -58,10 +79,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"))
@@ -86,7 +111,7 @@ lazy val metrics = (project in file("metrics"))
     libraryDependencies ++= Seq(
       smetrics,
       Testing.munit % Test,
-    )
+    ),
   )
 
 lazy val `persistence-cassandra` = (project in file("persistence-cassandra"))
@@ -98,6 +123,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,13 +176,16 @@ lazy val journal = (project in file("kafka-journal"))
       KafkaJournal.journal,
       KafkaJournal.persistence,
       Testing.munit % Test,
-    )
+    ),
   )
+  .settings(crossScalaVersions -= Scala3Version)
 
 lazy val docs = (project in file("kafka-flow-docs"))
   .dependsOn(core, `persistence-cassandra`, `persistence-kafka`, metrics)
   .settings(commonSettings)
   .enablePlugins(MdocPlugin, DocusaurusPlugin)
-  .settings(scalacOptions -= "-Xfatal-warnings")
+  .settings(
+    scalacOptions -= "-Xfatal-warnings",
+  )
 
 addCommandAlias("check", "versionPolicyCheck")

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)
     }

core/src/main/scala-3/com/evolutiongaming/kafka/flow/persistence/compression/CodecOps.scala

@@ -0,0 +1,16 @@
+package com.evolutiongaming.kafka.flow.persistence.compression
+
+import scodec.Codec
+import scodec.Codec.inlineImplementations
+
+// format: off
+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
+// format: on

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))
+      }(using pickMinOffset))
     }
 
     def offsetToCommit(getMinOffset: F[Option[Offset]]): F[Option[Offset]] = for {
@@ -249,7 +249,7 @@ object PartitionFlow {
 
       allowedOffset = minimumOffset getOrElse maximumOffset
 
-      // we move forward if minimum offset became larger or it is empty,
+      // we move forward if minimum offset became larger, or it is empty,
       // i.e. if we dealt with all the states, and there is nothing holding
       // us from moving forward
       committedOffsetValue <- committedOffset.get

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 {

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"))
 

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
     }

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,

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)