chore: Rewrite to scala3 syntax

.scalafmt.conf

@@ -1,5 +1,5 @@
 version                                  = 3.11.1
-runner.dialect                           = scala213
+runner.dialect                           = scala213source3
 project.git                              = true
 style                                    = defaultWithAlign
 docstrings.style                         = Asterisk
@@ -75,3 +75,13 @@ project.excludeFilters = [
   "scripts/authors.scala"
 ]
 project.layout         = StandardConvention
+
+rewrite.scala3.convertToNewSyntax = true
+runner {
+  dialectOverride {
+    allowSignificantIndentation = false
+    allowAsForImportRename = false
+    allowStarWildcardImport = false
+    allowPostfixStarVarargSplices = false
+  }
+}

cassandra-test/src/test/scala/org/apache/pekko/projection/cassandra/CassandraProjectionSpec.scala

@@ -189,7 +189,7 @@ class CassandraProjectionSpec
   private def genRandomProjectionId() =
     ProjectionId(UUID.randomUUID().toString, UUID.randomUUID().toString)
 
-  @tailrec private def eventuallyExpectError(sinkProbe: TestSubscriber.Probe[_]): Throwable = {
+  @tailrec private def eventuallyExpectError(sinkProbe: TestSubscriber.Probe[?]): Throwable = {
     sinkProbe.expectNextOrError() match {
       case Right(_)  => eventuallyExpectError(sinkProbe)
       case Left(exc) => exc

cassandra/src/main/scala/org/apache/pekko/projection/cassandra/internal/CassandraOffsetStore.scala

@@ -36,7 +36,7 @@ import com.datastax.oss.driver.api.core.cql.Statement
 /**
  * INTERNAL API
  */
-@InternalApi private[projection] class CassandraOffsetStore(system: ActorSystem[_], clock: Clock) {
+@InternalApi private[projection] class CassandraOffsetStore(system: ActorSystem[?], clock: Clock) {
   private val offsetSerialization = new OffsetSerialization(system)
   import offsetSerialization.fromStorageRepresentation
   import offsetSerialization.toStorageRepresentation
@@ -51,7 +51,7 @@ import com.datastax.oss.driver.api.core.cql.Statement
   val managementTable: String = cassandraSettings.managementTable
   private val cassandraPartitions = 5
 
-  def this(system: ActorSystem[_]) =
+  def this(system: ActorSystem[?]) =
     this(system, Clock.systemUTC())
 
   private def selectOne[T <: Statement[T]](stmt: Statement[T]): Future[Option[Row]] = {
@@ -79,7 +79,7 @@ import com.datastax.oss.driver.api.core.cql.Statement
     // same time let us query all rows for a single projection_name easily
     val partition = idToPartition(projectionId)
     offset match {
-      case _: MergeableOffset[_] =>
+      case _: MergeableOffset[?] =>
         throw new IllegalArgumentException("The CassandraOffsetStore does not currently support MergeableOffset")
       case _ =>
         val SingleOffset(_, manifest, offsetStr, _) =

cassandra/src/main/scala/org/apache/pekko/projection/cassandra/internal/CassandraProjectionImpl.scala

@@ -87,7 +87,7 @@ import pekko.stream.scaladsl.Source
   /*
    * Build the final ProjectionSettings to use, if currently set to None fallback to values in config file
    */
-  private def settingsOrDefaults(implicit system: ActorSystem[_]): ProjectionSettings = {
+  private def settingsOrDefaults(implicit system: ActorSystem[?]): ProjectionSettings = {
     val settings = settingsOpt.getOrElse(ProjectionSettings(system))
     restartBackoffOpt match {
       case None    => settings
@@ -158,7 +158,7 @@ import pekko.stream.scaladsl.Source
    * Return a RunningProjection
    */
   @InternalApi
-  override private[projection] def run()(implicit system: ActorSystem[_]): RunningProjection = {
+  override private[projection] def run()(implicit system: ActorSystem[?]): RunningProjection = {
     new CassandraInternalProjectionState(settingsOrDefaults).newRunningInstance()
   }
 
@@ -169,7 +169,7 @@ import pekko.stream.scaladsl.Source
    * This is mainly intended to be used by the TestKit allowing it to attach a TestSink to it.
    */
   @InternalApi
-  override private[projection] def mappedSource()(implicit system: ActorSystem[_]): Source[Done, Future[Done]] = {
+  override private[projection] def mappedSource()(implicit system: ActorSystem[?]): Source[Done, Future[Done]] = {
     new CassandraInternalProjectionState(settingsOrDefaults).mappedSource()
   }
 
@@ -178,7 +178,7 @@ import pekko.stream.scaladsl.Source
    * This internal class will hold the KillSwitch that is needed
    * when building the mappedSource and when running the projection (to stop)
    */
-  private class CassandraInternalProjectionState(val settings: ProjectionSettings)(implicit val system: ActorSystem[_])
+  private class CassandraInternalProjectionState(val settings: ProjectionSettings)(implicit val system: ActorSystem[?])
       extends InternalProjectionState[Offset, Envelope](
         projectionId,
         sourceProvider,
@@ -210,9 +210,9 @@ import pekko.stream.scaladsl.Source
   }
 
   private class CassandraRunningProjection(
-      source: Source[Done, _],
+      source: Source[Done, ?],
       offsetStore: CassandraOffsetStore,
-      projectionState: CassandraInternalProjectionState)(implicit system: ActorSystem[_])
+      projectionState: CassandraInternalProjectionState)(implicit system: ActorSystem[?])
       extends RunningProjection
       with RunningProjectionManagement[Offset] {
 

cassandra/src/main/scala/org/apache/pekko/projection/cassandra/internal/CassandraSettings.scala

@@ -36,6 +36,6 @@ private[projection] case class CassandraSettings(config: Config) {
 @InternalApi
 private[projection] object CassandraSettings {
 
-  def apply(system: ActorSystem[_]): CassandraSettings =
+  def apply(system: ActorSystem[?]): CassandraSettings =
     CassandraSettings(system.settings.config.getConfig("pekko.projection.cassandra"))
 }

cassandra/src/main/scala/org/apache/pekko/projection/cassandra/javadsl/CassandraProjection.scala

@@ -120,7 +120,7 @@ object CassandraProjection {
   def atLeastOnceFlow[Offset, Envelope](
       projectionId: ProjectionId,
       sourceProvider: SourceProvider[Offset, Envelope],
-      handler: FlowWithContext[Envelope, ProjectionContext, Done, ProjectionContext, _])
+      handler: FlowWithContext[Envelope, ProjectionContext, Done, ProjectionContext, ?])
       : AtLeastOnceFlowProjection[Offset, Envelope] =
     new CassandraProjectionImpl(
       projectionId,
@@ -154,7 +154,7 @@ object CassandraProjection {
    * For production it's recommended to create the table with DDL statements
    * before the system is started.
    */
-  def createTablesIfNotExists(system: ActorSystem[_]): CompletionStage[Done] = {
+  def createTablesIfNotExists(system: ActorSystem[?]): CompletionStage[Done] = {
     import scala.jdk.FutureConverters._
     val offsetStore = new CassandraOffsetStore(system)
     offsetStore.createKeyspaceAndTable().asJava

cassandra/src/main/scala/org/apache/pekko/projection/cassandra/scaladsl/CassandraProjection.scala

@@ -122,7 +122,7 @@ object CassandraProjection {
   def atLeastOnceFlow[Offset, Envelope](
       projectionId: ProjectionId,
       sourceProvider: SourceProvider[Offset, Envelope],
-      handler: FlowWithContext[Envelope, ProjectionContext, Done, ProjectionContext, _])
+      handler: FlowWithContext[Envelope, ProjectionContext, Done, ProjectionContext, ?])
       : AtLeastOnceFlowProjection[Offset, Envelope] =
     new CassandraProjectionImpl(
       projectionId,
@@ -156,7 +156,7 @@ object CassandraProjection {
    * For production, it's recommended to create the table with DDL statements
    * before the system is started.
    */
-  def createTablesIfNotExists()(implicit system: ActorSystem[_]): Future[Done] = {
+  def createTablesIfNotExists()(implicit system: ActorSystem[?]): Future[Done] = {
     val offsetStore = new CassandraOffsetStore(system)
     offsetStore.createKeyspaceAndTable()
   }

core-test/src/test/scala/org/apache/pekko/projection/ProjectionBehaviorSpec.scala

@@ -101,7 +101,7 @@ object ProjectionBehaviorSpec {
         () => new TestInMemoryOffsetStoreImpl[Int](),
         None) {
 
-    override private[projection] def newState(implicit system: ActorSystem[_]): TestInternalProjectionState[Int, Int] =
+    override private[projection] def newState(implicit system: ActorSystem[?]): TestInternalProjectionState[Int, Int] =
       new ProjectionBehaviourTestInternalProjectionState(
         projectionId,
         sourceProvider,
@@ -120,7 +120,7 @@ object ProjectionBehaviorSpec {
         statusObserver: StatusObserver[Int],
         offsetStore: TestOffsetStore[Int],
         testProbe: TestProbe[ProbeMessage],
-        failToStop: Boolean)(implicit system: ActorSystem[_])
+        failToStop: Boolean)(implicit system: ActorSystem[?])
         extends TestInternalProjectionState[Int, Int](
           projectionId,
           sourceProvider,
@@ -141,11 +141,11 @@ object ProjectionBehaviorSpec {
 
     private[projection] class ProjectionBehaviourTestRunningProjection(
         projectionId: ProjectionId,
-        source: Source[Done, _],
+        source: Source[Done, ?],
         killSwitch: SharedKillSwitch,
         offsetStore: TestOffsetStore[Int],
         testProbe: TestProbe[ProbeMessage],
-        failToStop: Boolean)(implicit _system: ActorSystem[_])
+        failToStop: Boolean)(implicit _system: ActorSystem[?])
         extends TestRunningProjection(source, killSwitch)
         with RunningProjectionManagement[Int] {
       import system.executionContext

core-test/src/test/scala/org/apache/pekko/projection/internal/TelemetryProviderSpec.scala

@@ -86,7 +86,7 @@ class TelemetryProviderEnsembleSpec extends ScalaTestWithActorTestKit(s"""
 object FakeTelemetry {
   val state = new AtomicReference[String]("")
 }
-class FakeTelemetry(projectionId: ProjectionId, system: ActorSystem[_]) extends Telemetry {
+class FakeTelemetry(projectionId: ProjectionId, system: ActorSystem[?]) extends Telemetry {
 
   override def failed(cause: Throwable): Unit = {}
 

core-test/src/test/scala/org/apache/pekko/projection/internal/metrics/tools/InMemTelemetry.scala

@@ -30,7 +30,7 @@ import pekko.projection.internal.Telemetry
 
 /**
  */
-class InMemTelemetry(projectionId: ProjectionId, system: ActorSystem[_]) extends Telemetry {
+class InMemTelemetry(projectionId: ProjectionId, system: ActorSystem[?]) extends Telemetry {
   private val instruments: InMemInstruments = InMemInstrumentsRegistry(system).forId(projectionId)
 
   import instruments._
@@ -74,9 +74,9 @@ case class ExternalContext(readyTimestampNanos: Long = System.nanoTime())
 case object TelemetryException extends RuntimeException("Oh, no! Handler errored.") with NoStackTrace
 
 object InMemInstrumentsRegistry extends ExtensionId[InMemInstrumentsRegistry] {
-  override def createExtension(system: ActorSystem[_]): InMemInstrumentsRegistry = new InMemInstrumentsRegistry(system)
+  override def createExtension(system: ActorSystem[?]): InMemInstrumentsRegistry = new InMemInstrumentsRegistry(system)
 }
-class InMemInstrumentsRegistry(system: ActorSystem[_]) extends Extension {
+class InMemInstrumentsRegistry(system: ActorSystem[?]) extends Extension {
   private val instrumentMap = new ConcurrentHashMap[ProjectionId, InMemInstruments]()
   def forId(projectionId: ProjectionId): InMemInstruments = {
     instrumentMap.computeIfAbsent(projectionId,
@@ -86,7 +86,7 @@ class InMemInstrumentsRegistry(system: ActorSystem[_]) extends Extension {
   }
 
   // these are added to use the constructor argument and keep the AkkaDisciplinePlugin happy
-  val observedActorSystem = new AtomicReference[ActorSystem[_]](null)
+  val observedActorSystem = new AtomicReference[ActorSystem[?]](null)
   observedActorSystem.set(system)
 }
 

core-test/src/test/scala/org/apache/pekko/projection/internal/metrics/tools/InternalProjectionStateMetricsSpec.scala

@@ -83,7 +83,7 @@ abstract class InternalProjectionStateMetricsSpec
 
   // inspired on ProjectionTestkit's runInternal
   protected def runInternal(
-      projectionState: InMemInternalProjectionState[_, _],
+      projectionState: InMemInternalProjectionState[?, ?],
       max: FiniteDuration = 3.seconds,
       interval: FiniteDuration = 100.millis)(assertFunction: => Unit): Unit = {
 
@@ -135,7 +135,7 @@ object InternalProjectionStateMetricsSpec {
       handlerStrategy: HandlerStrategy,
       numberOfEnvelopes: Int = 6,
       statusObserver: StatusObserver[Envelope] = NoopStatusObserver)(
-      implicit system: ActorSystem[_],
+      implicit system: ActorSystem[?],
       projectionId: ProjectionId) {
 
     private implicit val exCtx: ExecutionContext = system.executionContext
@@ -196,7 +196,7 @@ object InternalProjectionStateMetricsSpec {
       handlerStrategy: HandlerStrategy,
       statusObserver: StatusObserver[Env],
       settings: ProjectionSettings,
-      offsetStore: TestInMemoryOffsetStoreImpl[Offset])(implicit val system: ActorSystem[_])
+      offsetStore: TestInMemoryOffsetStoreImpl[Offset])(implicit val system: ActorSystem[?])
       extends InternalProjectionState[Offset, Env](
         projectionId,
         sourceProvider,
@@ -220,7 +220,7 @@ object InternalProjectionStateMetricsSpec {
     def newRunningInstance(): RunningProjection =
       new TestRunningProjection(RunningProjection.withBackoff(() => mappedSource(), settings), killSwitch)
 
-    class TestRunningProjection(val source: Source[Done, _], killSwitch: SharedKillSwitch) extends RunningProjection {
+    class TestRunningProjection(val source: Source[Done, ?], killSwitch: SharedKillSwitch) extends RunningProjection {
 
       private val futureDone = source.run()
 

core/src/main/scala/org/apache/pekko/projection/Projection.scala

@@ -83,7 +83,7 @@ trait Projection[Envelope] {
    * This is mainly intended to be used by the TestKit allowing it to attach a TestSink to it.
    */
   @InternalApi
-  private[projection] def mappedSource()(implicit system: ActorSystem[_]): Source[Done, Future[Done]]
+  private[projection] def mappedSource()(implicit system: ActorSystem[?]): Source[Done, Future[Done]]
 
   /**
    * INTERNAL API
@@ -95,7 +95,7 @@ trait Projection[Envelope] {
    * Return a RunningProjection
    */
   @InternalApi
-  private[projection] def run()(implicit system: ActorSystem[_]): RunningProjection
+  private[projection] def run()(implicit system: ActorSystem[?]): RunningProjection
 
 }
 
@@ -112,7 +112,7 @@ private[projection] object RunningProjection {
    */
   case object AbortProjectionException extends RuntimeException("Projection aborted.") with NoStackTrace
 
-  def withBackoff(source: () => Source[Done, _], settings: ProjectionSettings): Source[Done, _] = {
+  def withBackoff(source: () => Source[Done, ?], settings: ProjectionSettings): Source[Done, ?] = {
     RestartSource
       .onFailuresWithBackoff(settings.restartBackoff) { () =>
         source()

core/src/main/scala/org/apache/pekko/projection/ProjectionBehavior.scala

@@ -85,7 +85,7 @@ object ProjectionBehavior {
           ctx.log.debug2("Started actor handler [{}] for projection [{}]", ref, projection.projectionId)
         }
         val running = projection.run()(ctx.system)
-        if (running.isInstanceOf[RunningProjectionManagement[_]])
+        if (running.isInstanceOf[RunningProjectionManagement[?]])
           ProjectionManagement(ctx.system).register(projection.projectionId, ctx.self)
         new ProjectionBehavior[Any, Envelope](ctx, projection, stashBuffer).started(running)
       }
@@ -155,7 +155,7 @@ object ProjectionBehavior {
 
         case isPaused: IsPaused =>
           running match {
-            case mgmt: RunningProjectionManagement[_] =>
+            case mgmt: RunningProjectionManagement[?] =>
               if (isPaused.projectionId == projectionId) {
                 context.pipeToSelf(mgmt.getManagementState()) {
                   case Success(state) => GetManagementStateResult(state, isPaused.replyTo)
@@ -172,7 +172,7 @@ object ProjectionBehavior {
 
         case setPaused: SetPaused =>
           running match {
-            case mgmt: RunningProjectionManagement[_] =>
+            case mgmt: RunningProjectionManagement[?] =>
               if (setPaused.projectionId == projectionId) {
                 context.log.info2(
                   "Running state will be changed to [{}] for projection [{}].",
@@ -246,7 +246,7 @@ object ProjectionBehavior {
     Behaviors.same
   }
 
-  private def settingPaused(setPaused: SetPaused, mgmt: RunningProjectionManagement[_]): Behavior[Command] =
+  private def settingPaused(setPaused: SetPaused, mgmt: RunningProjectionManagement[?]): Behavior[Command] =
     Behaviors.receiveMessage {
       case Stopped =>
         context.log.debug("Projection [{}] stopped", projectionId)

core/src/main/scala/org/apache/pekko/projection/internal/HandlerRecoveryImpl.scala

@@ -66,7 +66,7 @@ import pekko.projection.StatusObserver
       lastOffset: Offset, // used for logging
       abort: Future[Done], // retries can be aborted by failing this Future
       futureCallback: () => Future[Done],
-      onSkip: () => Future[Done] = HandlerRecoveryImpl.defaultOnSkip)(implicit system: ActorSystem[_]): Future[Done] = {
+      onSkip: () => Future[Done] = HandlerRecoveryImpl.defaultOnSkip)(implicit system: ActorSystem[?]): Future[Done] = {
     import HandlerRecoveryStrategy.Internal._
 
     implicit val scheduler: Scheduler = system.classicSystem.scheduler

core/src/main/scala/org/apache/pekko/projection/internal/InternalProjectionState.scala

@@ -59,7 +59,7 @@ private[projection] abstract class InternalProjectionState[Offset, Envelope](
     settings: ProjectionSettings) {
 
   def logger: LoggingAdapter
-  implicit def system: ActorSystem[_]
+  implicit def system: ActorSystem[?]
   implicit def executionContext: ExecutionContext
 
   private var telemetry: Telemetry = NoopTelemetry
@@ -273,10 +273,10 @@ private[projection] abstract class InternalProjectionState[Offset, Envelope](
       }
 
       sourceProvider match {
-        case _: MergeableOffsetSourceProvider[_, _] =>
+        case _: MergeableOffsetSourceProvider[?, ?] =>
           val batches = envelopesAndOffsets
             .flatMap {
-              case context @ ProjectionContextImpl(offset: MergeableOffset[_] @unchecked, _, _, _) =>
+              case context @ ProjectionContextImpl(offset: MergeableOffset[?] @unchecked, _, _, _) =>
                 offset.entries.toSeq.map {
                   case (key, _) => (key, context)
                 }
@@ -360,7 +360,7 @@ private[projection] abstract class InternalProjectionState[Offset, Envelope](
               }
           }
 
-      case _: FlowHandlerStrategy[_] =>
+      case _: FlowHandlerStrategy[?] =>
         // not possible, no API for this
         throw new IllegalStateException("Unsupported combination of exactlyOnce and flow")
     }

core/src/main/scala/org/apache/pekko/projection/internal/OffsetSerialization.scala

@@ -46,7 +46,7 @@ import pekko.serialization.Serializers
 /**
  * INTERNAL API
  */
-@InternalApi private[projection] class OffsetSerialization(system: ActorSystem[_]) {
+@InternalApi private[projection] class OffsetSerialization(system: ActorSystem[?]) {
   import OffsetSerialization._
 
   private val serialization = SerializationExtension(system)
@@ -101,7 +101,7 @@ import pekko.serialization.Serializers
       case i: Int                   => SingleOffset(id, IntManifest, i.toString, mergeable)
       case seq: query.Sequence      => SingleOffset(id, SequenceManifest, seq.value.toString, mergeable)
       case tbu: query.TimeBasedUUID => SingleOffset(id, TimeBasedUUIDManifest, tbu.value.toString, mergeable)
-      case mrg: MergeableOffset[_]  =>
+      case mrg: MergeableOffset[?]  =>
         val list = mrg.entries.map {
           case (key, innerOffset) =>
             toStorageRepresentation(ProjectionId(id.name, key), innerOffset, mergeable = true)

core/src/main/scala/org/apache/pekko/projection/internal/OffsetStrategy.scala

@@ -138,7 +138,7 @@ private[projection] final case class GroupedHandlerStrategy[Envelope](
  */
 @InternalApi
 private[projection] final case class FlowHandlerStrategy[Envelope](
-    flowCtx: FlowWithContext[Envelope, ProjectionContext, Done, ProjectionContext, _])
+    flowCtx: FlowWithContext[Envelope, ProjectionContext, Done, ProjectionContext, ?])
     extends HandlerStrategy {
 
   override def recreateHandlerOnNextAccess(): Unit = ()