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

core/src/main/scala/org/apache/pekko/persistence/r2dbc/ConnectionFactoryProvider.scala

@@ -43,10 +43,10 @@ import io.r2dbc.spi.ConnectionFactoryOptions
 import io.r2dbc.spi.Option
 
 object ConnectionFactoryProvider extends ExtensionId[ConnectionFactoryProvider] {
-  def createExtension(system: ActorSystem[_]): ConnectionFactoryProvider = new ConnectionFactoryProvider(system)
+  def createExtension(system: ActorSystem[?]): ConnectionFactoryProvider = new ConnectionFactoryProvider(system)
 
   // Java API
-  def get(system: ActorSystem[_]): ConnectionFactoryProvider = apply(system)
+  def get(system: ActorSystem[?]): ConnectionFactoryProvider = apply(system)
 
   /**
    * Enables customization of [[ConnectionFactoryOptions]] right before the connection factory is created.
@@ -76,7 +76,7 @@ object ConnectionFactoryProvider extends ExtensionId[ConnectionFactoryProvider]
   }
 }
 
-class ConnectionFactoryProvider(system: ActorSystem[_]) extends Extension {
+class ConnectionFactoryProvider(system: ActorSystem[?]) extends Extension {
 
   private val sessions = new ConcurrentHashMap[String, ConnectionPool]
 
@@ -111,7 +111,7 @@ class ConnectionFactoryProvider(system: ActorSystem[_]) extends Extension {
     settings.connectionFactoryOptionsCustomizer match {
       case None       => NoopCustomizer
       case Some(fqcn) =>
-        val args = List(classOf[ActorSystem[_]] -> system)
+        val args = List(classOf[ActorSystem[?]] -> system)
         system.dynamicAccess.createInstanceFor[ConnectionFactoryOptionsCustomizer](fqcn, args) match {
           case Success(customizer) => customizer
           case Failure(cause)      =>

core/src/main/scala/org/apache/pekko/persistence/r2dbc/R2dbcSettings.scala

@@ -143,7 +143,7 @@ final class StateSettings(val config: Config) extends ConnectionSettings with Us
   @InternalApi private[pekko] val durableStateAdditionalColumnClasses: Map[String, immutable.IndexedSeq[String]] = {
     val cfg = config.getConfig("additional-columns")
     cfg.root.unwrapped.asScala.toMap.map {
-      case (k, v: java.util.List[_]) => k -> v.iterator.asScala.map(_.toString).toVector
+      case (k, v: java.util.List[?]) => k -> v.iterator.asScala.map(_.toString).toVector
       case (k, v)                    => k -> Vector(v.toString)
     }
   }

core/src/main/scala/org/apache/pekko/persistence/r2dbc/cleanup/scaladsl/DurableStateCleanup.scala

@@ -57,7 +57,7 @@ final class DurableStateCleanup(systemProvider: ClassicActorSystemProvider, conf
   /**
    * INTERNAL API
    */
-  @InternalApi private[pekko] implicit val system: ActorSystem[_] = {
+  @InternalApi private[pekko] implicit val system: ActorSystem[?] = {
     import pekko.actor.typed.scaladsl.adapter._
     systemProvider.classicSystem.toTyped
   }

core/src/main/scala/org/apache/pekko/persistence/r2dbc/cleanup/scaladsl/EventSourcedCleanup.scala

@@ -63,7 +63,7 @@ final class EventSourcedCleanup(systemProvider: ClassicActorSystemProvider, conf
   /**
    * INTERNAL API
    */
-  @InternalApi private[pekko] implicit val system: ActorSystem[_] = {
+  @InternalApi private[pekko] implicit val system: ActorSystem[?] = {
     import pekko.actor.typed.scaladsl.adapter._
     systemProvider.classicSystem.toTyped
   }

core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/AdditionalColumnFactory.scala

@@ -34,7 +34,7 @@ import pekko.persistence.r2dbc.state.{ javadsl => javadslState }
   final class AdditionalColumnAdapter(delegate: javadslState.AdditionalColumn[Any, Any])
       extends AdditionalColumn[Any, Any] {
 
-    override private[pekko] val fieldClass: Class[_] =
+    override private[pekko] val fieldClass: Class[?] =
       delegate.fieldClass
 
     override def columnName: String =
@@ -48,23 +48,23 @@ import pekko.persistence.r2dbc.state.{ javadsl => javadslState }
         upsert.revision,
         upsert.value)
       delegate.bind(javadslUpsert) match {
-        case bindValue: javadslState.AdditionalColumn.BindValue[_] => AdditionalColumn.BindValue(bindValue.value)
+        case bindValue: javadslState.AdditionalColumn.BindValue[?] => AdditionalColumn.BindValue(bindValue.value)
         case javadslState.AdditionalColumn.BindNull                => AdditionalColumn.BindNull
         case javadslState.AdditionalColumn.Skip                    => AdditionalColumn.Skip
       }
     }
 
   }
 
-  def create(system: ActorSystem[_], fqcn: String): AdditionalColumn[Any, Any] = {
+  def create(system: ActorSystem[?], fqcn: String): AdditionalColumn[Any, Any] = {
     val dynamicAccess = system.classicSystem.asInstanceOf[ExtendedActorSystem].dynamicAccess
 
     def tryCreateScaladslInstance(): Try[AdditionalColumn[Any, Any]] = {
       dynamicAccess
         .createInstanceFor[AdditionalColumn[Any, Any]](fqcn, Nil)
         .orElse(
           dynamicAccess
-            .createInstanceFor[AdditionalColumn[Any, Any]](fqcn, List(classOf[ActorSystem[_]] -> system))
+            .createInstanceFor[AdditionalColumn[Any, Any]](fqcn, List(classOf[ActorSystem[?]] -> system))
             .orElse(dynamicAccess.createInstanceFor[AdditionalColumn[Any, Any]](
               fqcn,
               List(classOf[ClassicActorSystem] -> system.classicSystem))))
@@ -77,7 +77,7 @@ import pekko.persistence.r2dbc.state.{ javadsl => javadslState }
           dynamicAccess
             .createInstanceFor[javadslState.AdditionalColumn[Any, Any]](
               fqcn,
-              List(classOf[ActorSystem[_]] -> system))
+              List(classOf[ActorSystem[?]] -> system))
             .orElse(dynamicAccess.createInstanceFor[javadslState.AdditionalColumn[Any, Any]](
               fqcn,
               List(classOf[ClassicActorSystem] -> system.classicSystem))))
@@ -91,7 +91,7 @@ import pekko.persistence.r2dbc.state.{ javadsl => javadslState }
       .getOrElse(
         throw new IllegalArgumentException(
           s"Additional column [$fqcn] must implement " +
-          s"[${classOf[AdditionalColumn[_, _]].getName}] or [${classOf[javadslState.AdditionalColumn[_, _]].getName}]. It " +
+          s"[${classOf[AdditionalColumn[?, ?]].getName}] or [${classOf[javadslState.AdditionalColumn[?, ?]].getName}]. It " +
           s"may have an ActorSystem constructor parameter."))
   }
 

core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/ChangeHandlerFactory.scala

@@ -43,15 +43,15 @@ import pekko.persistence.r2dbc.state.scaladsl.ChangeHandler
     }
   }
 
-  def create(system: ActorSystem[_], fqcn: String): ChangeHandler[Any] = {
+  def create(system: ActorSystem[?], fqcn: String): ChangeHandler[Any] = {
     val dynamicAccess = system.classicSystem.asInstanceOf[ExtendedActorSystem].dynamicAccess
 
     def tryCreateScaladslInstance(): Try[ChangeHandler[Any]] = {
       dynamicAccess
         .createInstanceFor[ChangeHandler[Any]](fqcn, Nil)
         .orElse(
           dynamicAccess
-            .createInstanceFor[ChangeHandler[Any]](fqcn, List(classOf[ActorSystem[_]] -> system))
+            .createInstanceFor[ChangeHandler[Any]](fqcn, List(classOf[ActorSystem[?]] -> system))
             .orElse(
               dynamicAccess
                 .createInstanceFor[ChangeHandler[Any]](
@@ -64,7 +64,7 @@ import pekko.persistence.r2dbc.state.scaladsl.ChangeHandler
         .createInstanceFor[javadslState.ChangeHandler[Any]](fqcn, Nil)
         .orElse(
           dynamicAccess
-            .createInstanceFor[javadslState.ChangeHandler[Any]](fqcn, List(classOf[ActorSystem[_]] -> system))
+            .createInstanceFor[javadslState.ChangeHandler[Any]](fqcn, List(classOf[ActorSystem[?]] -> system))
             .orElse(
               dynamicAccess
                 .createInstanceFor[javadslState.ChangeHandler[Any]](
@@ -80,7 +80,7 @@ import pekko.persistence.r2dbc.state.scaladsl.ChangeHandler
       .getOrElse(
         throw new IllegalArgumentException(
           s"Change handler [$fqcn] must implement " +
-          s"[${classOf[ChangeHandler[_]].getName}] or [${classOf[javadslState.ChangeHandler[_]].getName}]. It " +
+          s"[${classOf[ChangeHandler[?]].getName}] or [${classOf[javadslState.ChangeHandler[?]].getName}]. It " +
           s"may have an ActorSystem constructor parameter."))
 
   }

core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/EnvelopeOrigin.scala

@@ -25,18 +25,18 @@ import pekko.persistence.query.typed.EventEnvelope
   val SourceBacktracking = "BT"
   val SourcePubSub = "PS"
 
-  def fromQuery(env: EventEnvelope[_]): Boolean =
+  def fromQuery(env: EventEnvelope[?]): Boolean =
     env.source == SourceQuery
 
-  def fromBacktracking(env: EventEnvelope[_]): Boolean =
+  def fromBacktracking(env: EventEnvelope[?]): Boolean =
     env.source == SourceBacktracking
 
-  def fromPubSub(env: EventEnvelope[_]): Boolean =
+  def fromPubSub(env: EventEnvelope[?]): Boolean =
     env.source == SourcePubSub
 
   def isFilteredEvent(env: Any): Boolean =
     env match {
-      case e: EventEnvelope[_] => e.filtered
+      case e: EventEnvelope[?] => e.filtered
       case _                   => false
     }
 }

core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/PayloadCodec.scala

@@ -24,7 +24,7 @@ import io.r2dbc.spi.Statement
  * INTERNAL API
  */
 @InternalApi private[pekko] sealed trait PayloadCodec {
-  def payloadClass: Class[_]
+  def payloadClass: Class[?]
   def encode(bytes: Array[Byte]): Any
   def decode(payload: Any): Array[Byte]
   def nonePayload: Array[Byte]

core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/PubSub.scala

@@ -41,17 +41,17 @@ import org.slf4j.LoggerFactory
 @InternalApi private[pekko] object PubSub extends ExtensionId[PubSub] {
   private val log = LoggerFactory.getLogger(classOf[PubSub])
 
-  def createExtension(system: ActorSystem[_]): PubSub = new PubSub(system)
+  def createExtension(system: ActorSystem[?]): PubSub = new PubSub(system)
 
   // Java API
-  def get(system: ActorSystem[_]): PubSub = apply(system)
+  def get(system: ActorSystem[?]): PubSub = apply(system)
 
 }
 
 /**
  * INTERNAL API
  */
-@InternalApi private[pekko] class PubSub(system: ActorSystem[_]) extends Extension {
+@InternalApi private[pekko] class PubSub(system: ActorSystem[?]) extends Extension {
   import PubSub.log
 
   private val topics = new ConcurrentHashMap[String, ActorRef[Any]]

core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutor.scala

@@ -83,14 +83,14 @@ import reactor.core.publisher.Mono
   def selectOneInTx[A](statement: Statement, mapRow: Row => A)(
       implicit
       ec: ExecutionContext,
-      system: ActorSystem[_]): Future[Option[A]] = {
+      system: ActorSystem[?]): Future[Option[A]] = {
     selectInTx(statement, mapRow).map(_.headOption)
   }
 
   def selectInTx[A](statement: Statement, mapRow: Row => A)(
       implicit
       ec: ExecutionContext,
-      system: ActorSystem[_]): Future[immutable.IndexedSeq[A]] = {
+      system: ActorSystem[?]): Future[immutable.IndexedSeq[A]] = {
     statement.execute().asFuture().flatMap { result =>
       val consumer: BiConsumer[mutable.Builder[A, immutable.IndexedSeq[A]], A] = (builder, elem) => builder += elem
       Flux
@@ -111,7 +111,7 @@ import reactor.core.publisher.Mono
 class R2dbcExecutor(val connectionFactory: ConnectionFactory, log: Logger, logDbCallsExceeding: FiniteDuration)(
     implicit
     ec: ExecutionContext,
-    system: ActorSystem[_]) {
+    system: ActorSystem[?]) {
   import R2dbcExecutor._
 
   private val logDbCallsExceedingMicros = logDbCallsExceeding.toMicros

core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/JournalDao.scala

@@ -81,7 +81,7 @@ private[r2dbc] object JournalDao {
   def fromConfig(
       settings: JournalSettings,
       config: Config
-  )(implicit system: ActorSystem[_], ec: ExecutionContext): JournalDao = {
+  )(implicit system: ActorSystem[?], ec: ExecutionContext): JournalDao = {
     val connectionFactory =
       ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory, config)
     settings.dialect match {
@@ -100,7 +100,7 @@ private[r2dbc] object JournalDao {
  */
 @InternalApi
 private[r2dbc] class JournalDao(val settings: JournalSettings, connectionFactory: ConnectionFactory)(
-    implicit val ec: ExecutionContext, system: ActorSystem[_]) extends EventsByPersistenceIdDao
+    implicit val ec: ExecutionContext, system: ActorSystem[?]) extends EventsByPersistenceIdDao
     with HighestSequenceNrDao {
   import JournalDao.SerializedJournalRow
   import JournalDao.log

core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/R2dbcJournal.scala

@@ -48,7 +48,7 @@ import pekko.stream.scaladsl.Sink
  */
 @InternalApi
 private[r2dbc] object R2dbcJournal {
-  case class WriteFinished(persistenceId: String, done: Future[_])
+  case class WriteFinished(persistenceId: String, done: Future[?])
 
   def deserializeRow(serialization: Serialization, row: SerializedJournalRow): PersistentRepr = {
     if (row.payload.isEmpty)
@@ -80,7 +80,7 @@ private[r2dbc] final class R2dbcJournal(config: Config, cfgPath: String) extends
   import R2dbcJournal.WriteFinished
   import R2dbcJournal.deserializeRow
 
-  implicit val system: ActorSystem[_] = context.system.toTyped
+  implicit val system: ActorSystem[?] = context.system.toTyped
   implicit val ec: ExecutionContext = context.dispatcher
 
   private val log = Logging(context.system, classOf[R2dbcJournal])
@@ -98,7 +98,7 @@ private[r2dbc] final class R2dbcJournal(config: Config, cfgPath: String) extends
 
   // if there are pending writes when an actor restarts we must wait for
   // them to complete before we can read the highest sequence number or we will miss it
-  private val writesInProgress = new java.util.HashMap[String, Future[_]]()
+  private val writesInProgress = new java.util.HashMap[String, Future[?]]()
 
   override def receivePluginInternal: Receive = { case WriteFinished(pid, f) =>
     writesInProgress.remove(pid, f)

core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/mysql/MySQLJournalDao.scala

@@ -119,7 +119,7 @@ private[r2dbc] object MySQLJournalDao {
 private[r2dbc] class MySQLJournalDao(
     journalSettings: JournalSettings,
     connectionFactory: ConnectionFactory)(
-    implicit ec: ExecutionContext, system: ActorSystem[_]
+    implicit ec: ExecutionContext, system: ActorSystem[?]
 ) extends JournalDao(journalSettings, connectionFactory) {
   MySQLJournalDao.settingRequirements(journalSettings)
 

core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/QueryDao.scala

@@ -58,7 +58,7 @@ object QueryDao {
   def fromConfig(
       settings: QuerySettings,
       config: Config
-  )(implicit system: ActorSystem[_], ec: ExecutionContext): QueryDao = {
+  )(implicit system: ActorSystem[?], ec: ExecutionContext): QueryDao = {
     val connectionFactory =
       ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory, config)
     settings.dialect match {
@@ -75,7 +75,7 @@ object QueryDao {
  */
 @InternalApi
 private[r2dbc] class QueryDao(val settings: QuerySettings, connectionFactory: ConnectionFactory)(
-    implicit val ec: ExecutionContext, system: ActorSystem[_]) extends BySliceQuery.Dao[SerializedJournalRow]
+    implicit val ec: ExecutionContext, system: ActorSystem[?]) extends BySliceQuery.Dao[SerializedJournalRow]
     with EventsByPersistenceIdDao with HighestSequenceNrDao {
   import JournalDao.readMetadata
   import QueryDao.log

core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala

@@ -81,7 +81,7 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
   private val log = LoggerFactory.getLogger(getClass)
   private val settings = QuerySettings(config)
 
-  private implicit val typedSystem: ActorSystem[_] = system.toTyped
+  private implicit val typedSystem: ActorSystem[?] = system.toTyped
   import typedSystem.executionContext
   private val serialization = SerializationExtension(system)
   private val persistenceExt = Persistence(system)

core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/mysql/MySQLQueryDao.scala

@@ -42,7 +42,7 @@ import io.r2dbc.spi.Row
 private[r2dbc] class MySQLQueryDao(
     querySettings: QuerySettings,
     connectionFactory: ConnectionFactory
-)(implicit ec: ExecutionContext, system: ActorSystem[_]) extends QueryDao(querySettings, connectionFactory) {
+)(implicit ec: ExecutionContext, system: ActorSystem[?]) extends QueryDao(querySettings, connectionFactory) {
 
   override lazy val statementTimestampSql: String = "NOW(6)"
 

core/src/main/scala/org/apache/pekko/persistence/r2dbc/session/javadsl/R2dbcSession.scala

@@ -38,12 +38,12 @@ object R2dbcSession {
    * Runs the passed function using a R2dbcSession with a new transaction. The connection is closed and the transaction
    * is committed at the end or rolled back in case of failures.
    */
-  def withSession[A](system: ActorSystem[_], fun: JFunction[R2dbcSession, CompletionStage[A]]): CompletionStage[A] = {
+  def withSession[A](system: ActorSystem[?], fun: JFunction[R2dbcSession, CompletionStage[A]]): CompletionStage[A] = {
     withSession(system, "pekko.persistence.r2dbc.connection-factory", fun)
   }
 
   def withSession[A](
-      system: ActorSystem[_],
+      system: ActorSystem[?],
       connectionFactoryConfigPath: String,
       fun: JFunction[R2dbcSession, CompletionStage[A]]): CompletionStage[A] = {
     scaladslSession.R2dbcSession.withSession(system, connectionFactoryConfigPath) { scaladslSession =>
@@ -55,7 +55,7 @@ object R2dbcSession {
 }
 
 @ApiMayChange
-final class R2dbcSession(val connection: Connection)(implicit ec: ExecutionContext, system: ActorSystem[_]) {
+final class R2dbcSession(val connection: Connection)(implicit ec: ExecutionContext, system: ActorSystem[?]) {
 
   def createStatement(sql: String): Statement =
     connection.createStatement(sql)

core/src/main/scala/org/apache/pekko/persistence/r2dbc/session/scaladsl/R2dbcSession.scala

@@ -38,11 +38,11 @@ object R2dbcSession {
    * Runs the passed function using a R2dbcSession with a new transaction. The connection is closed and the transaction
    * is committed at the end or rolled back in case of failures.
    */
-  def withSession[A](system: ActorSystem[_])(fun: R2dbcSession => Future[A]): Future[A] = {
+  def withSession[A](system: ActorSystem[?])(fun: R2dbcSession => Future[A]): Future[A] = {
     withSession(system, "pekko.persistence.r2dbc.connection-factory")(fun)
   }
 
-  def withSession[A](system: ActorSystem[_], connectionFactoryConfigPath: String)(
+  def withSession[A](system: ActorSystem[?], connectionFactoryConfigPath: String)(
       fun: R2dbcSession => Future[A]): Future[A] = {
     val connectionFactory = ConnectionFactoryProvider(system).connectionFactoryFor(connectionFactoryConfigPath)
     val r2dbcExecutor = new R2dbcExecutor(connectionFactory, log, logDbCallsDisabled)(system.executionContext, system)
@@ -54,7 +54,7 @@ object R2dbcSession {
 }
 
 @ApiMayChange
-final class R2dbcSession(val connection: Connection)(implicit val ec: ExecutionContext, val system: ActorSystem[_]) {
+final class R2dbcSession(val connection: Connection)(implicit val ec: ExecutionContext, val system: ActorSystem[?]) {
 
   def createStatement(sql: String): Statement =
     connection.createStatement(sql)

core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/R2dbcSnapshotStore.scala

@@ -57,7 +57,7 @@ private[r2dbc] final class R2dbcSnapshotStore(cfg: Config, cfgPath: String) exte
 
   private implicit val ec: ExecutionContext = context.dispatcher
   private val serialization: Serialization = SerializationExtension(context.system)
-  private implicit val system: ActorSystem[_] = context.system.toTyped
+  private implicit val system: ActorSystem[?] = context.system.toTyped
 
   private val dao = {
     val settings = SnapshotSettings(cfg)