Fix migration warnings in scio-core

build.sbt

@@ -471,9 +471,6 @@ lazy val `scio-core`: Project = project
         )
       else Nil
     },
-    scalacOptions ++= {
-      if (scalaVersion.value.startsWith("3")) Seq("-source:3.0-migration") else Nil
-    },
     compileOrder := CompileOrder.JavaThenScala,
   )
   .dependsOn(

scio-core/src/main/scala/com/spotify/scio/ScioContext.scala

@@ -167,7 +167,7 @@ object ContextAndArgs {
     def parse(args: Array[String]): F[Result]
   }
 
-  final case class DefaultParser[T <: PipelineOptions: ClassTag] private ()
+  final case class DefaultParser[T <: PipelineOptions: ClassTag] private[scio] ()
       extends ArgsParser[Try] {
     override type ArgsType = Args
 
@@ -213,7 +213,7 @@ object ContextAndArgs {
           case _ => true
         }
 
-      CaseApp.detailedParseWithHelp[T](customArgs) match {
+      CaseApp.detailedParseWithHelp[T](customArgs.toIndexedSeq) match {
         case Left(error) =>
           Failure(new Exception(error.message))
         case Right((_, _, help, _)) if help =>
@@ -245,7 +245,7 @@ object ContextAndArgs {
     }
   }
 
-  final case class PipelineOptionsParser[T <: PipelineOptions: ClassTag] private ()
+  final case class PipelineOptionsParser[T <: PipelineOptions: ClassTag] private[scio] ()
       extends ArgsParser[Try] {
     override type ArgsType = T
 
@@ -412,7 +412,7 @@ object ScioContext {
         }
     } yield s"--$str($$|=)".r
 
-    val patterns = registeredPatterns + "--help($$|=)".r
+    val patterns = registeredPatterns.union(Set("--help($$|=)".r))
 
     // Split cmdlineArgs into 2 parts, optArgs for PipelineOptions and appArgs for Args
     val (optArgs, appArgs) =
@@ -690,7 +690,7 @@ class ScioContext private[scio] (
               BuildInfo.version,
               BuildInfo.scalaVersion,
               sc.optionsAs[ApplicationNameOptions].getAppName,
-              state.toString,
+              this.state.toString,
               getBeamMetrics
             )
 

scio-core/src/main/scala/com/spotify/scio/coders/Coder.scala

@@ -76,19 +76,20 @@ private[scio] object Ref {
   def unapply[T](c: Ref[T]): Option[(String, Coder[T])] = Option((c.typeName, c.value))
 }
 
-final case class RawBeam[T] private (beam: BCoder[T]) extends Coder[T] {
+final case class RawBeam[T] private[coders] (beam: BCoder[T]) extends Coder[T] {
   override def toString: String = s"RawBeam($beam)"
 }
-final case class Beam[T] private (beam: BCoder[T]) extends Coder[T] {
+final case class Beam[T] private[coders] (beam: BCoder[T]) extends Coder[T] {
   override def toString: String = s"Beam($beam)"
 }
-final case class Fallback[T] private (ct: ClassTag[T]) extends Coder[T] {
+final case class Fallback[T] private[coders] (ct: ClassTag[T]) extends Coder[T] {
   override def toString: String = s"Fallback($ct)"
 }
-final case class Transform[A, B] private (c: Coder[A], f: BCoder[A] => Coder[B]) extends Coder[B] {
+final case class Transform[A, B] private[coders] (c: Coder[A], f: BCoder[A] => Coder[B])
+    extends Coder[B] {
   override def toString: String = s"Transform($c, $f)"
 }
-final case class Disjunction[T, Id] private (
+final case class Disjunction[T, Id] private[coders] (
   typeName: String,
   idCoder: Coder[Id],
   id: T => Id,
@@ -97,7 +98,7 @@ final case class Disjunction[T, Id] private (
   override def toString: String = s"Disjunction($typeName, $coder)"
 }
 
-final case class Record[T] private (
+final case class Record[T] private[coders] (
   typeName: String,
   cs: Array[(String, Coder[Any])],
   construct: Seq[Any] => T,
@@ -112,7 +113,8 @@ final case class Record[T] private (
 }
 
 // KV are special in beam and need to be serialized using an instance of KvCoder.
-final case class KVCoder[K, V] private (koder: Coder[K], voder: Coder[V]) extends Coder[KV[K, V]] {
+final case class KVCoder[K, V] private[coders] (koder: Coder[K], voder: Coder[V])
+    extends Coder[KV[K, V]] {
   override def toString: String = s"KVCoder($koder, $voder)"
 }
 

scio-core/src/main/scala/com/spotify/scio/io/Tap.scala

@@ -120,7 +120,7 @@ object MaterializeTap {
     new MaterializeTap(path, CoderMaterializer.beam(context, Coder[T]))
 }
 
-final case class ClosedTap[T] private (private[scio] val underlying: Tap[T]) {
+final case class ClosedTap[T] private[scio] (private[scio] val underlying: Tap[T]) {
 
   /**
    * Get access to the underlying Tap. The ScioContext has to be ran before.

scio-core/src/main/scala/com/spotify/scio/io/Taps.scala

@@ -99,7 +99,9 @@ final private class PollingTaps(private[this] val backOff: BackOff) extends Taps
         import scala.concurrent.ExecutionContext.Implicits.global
         Future {
           val sleeper = Sleeper.DEFAULT
-          do {
+          var first = true
+          while (first || BackOffUtils.next(sleeper, backOff)) {
+            first = false
             if (polls.nonEmpty) {
               val tap = if (polls.size > 1) "taps" else "tap"
               logger.info(s"Polling for ${polls.size} $tap")
@@ -112,7 +114,7 @@ final private class PollingTaps(private[this] val backOff: BackOff) extends Taps
               }
               polls = pending
             }
-          } while (BackOffUtils.next(sleeper, backOff))
+          }
           polls.foreach(p => p.promise.failure(new TapNotAvailableException(p.name)))
         }
       }

scio-core/src/main/scala/com/spotify/scio/schemas/Schema.scala

@@ -142,7 +142,7 @@ object LogicalType {
     Some(logicalType.underlying)
 }
 
-final case class Record[T] private (
+final case class Record[T] private[schemas] (
   schemas: Array[(String, Schema[Any])],
   construct: Seq[Any] => T,
   destruct: T => Array[Any]

scio-core/src/main/scala/com/spotify/scio/util/ProtobufUtil.scala

@@ -46,6 +46,7 @@ object ProtobufUtil {
    */
   def schemaMetadataOf[T <: Message: ClassTag]: Map[String, AnyRef] = {
     import me.lyh.protobuf.generic
+    import me.lyh.protobuf.generic.JsonSchema
     val schema = generic.Schema
       .of[Message](classTag[T].asInstanceOf[ClassTag[Message]])
       .toJson