bump: update akka to 2.10.0-M1 (#84)

akka · Oct 15, 2024 · 0537a73 · 0537a73
1 parent fcb55ac
commit 0537a73
Show file tree

Hide file tree

Showing 23 changed files with 94 additions and 113 deletions.
diff --git a/.jvmopts b/.jvmopts
diff --git a/core/src/main/scala/akka/persistence/dynamodb/cleanup/scaladsl/EventSourcedCleanup.scala b/core/src/main/scala/akka/persistence/dynamodb/cleanup/scaladsl/EventSourcedCleanup.scala
@@ -15,7 +15,6 @@ import scala.util.Success
 import akka.Done
 import akka.actor.ClassicActorSystemProvider
 import akka.actor.typed.ActorSystem
-import akka.actor.typed.scaladsl.LoggerOps
 import akka.annotation.ApiMayChange
 import akka.annotation.InternalApi
 import akka.persistence.SnapshotSelectionCriteria
@@ -549,7 +548,7 @@ final class EventSourcedCleanup(systemProvider: ClassicActorSystemProvider, conf
         case pid :: tail =>
           pidOperation(pid).flatMap { _ =>
             if (n % settings.cleanupSettings.logProgressEvery == 0)
-              log.infoN("Cleanup {} [{}] of [{}].", operationName, n, size)
+              log.info("Cleanup {} [{}] of [{}].", operationName, n, size)
             loop(tail, n + 1)
           }
       }
@@ -559,7 +558,7 @@ final class EventSourcedCleanup(systemProvider: ClassicActorSystemProvider, conf
 
     result.onComplete {
       case Success(_) =>
-        log.info2("Cleanup completed {} of [{}] persistenceId.", operationName, size)
+        log.info("Cleanup completed {} of [{}] persistenceId.", operationName, size)
       case Failure(e) =>
         log.error(s"Cleanup {$operationName} failed.", e)
     }

diff --git a/core/src/main/scala/akka/persistence/dynamodb/internal/BySliceQuery.scala b/core/src/main/scala/akka/persistence/dynamodb/internal/BySliceQuery.scala
@@ -11,7 +11,6 @@ import scala.concurrent.ExecutionContext
 import scala.concurrent.duration.FiniteDuration
 
 import akka.NotUsed
-import akka.actor.typed.scaladsl.LoggerOps
 import akka.annotation.InternalApi
 import akka.persistence.dynamodb.DynamoDBSettings
 import akka.persistence.query.Offset
@@ -118,7 +117,7 @@ import org.slf4j.Logger
         }
 
         if (state.queryCount != 0 && log.isDebugEnabled())
-          log.debugN(
+          log.debug(
             "{} next query [{}] from slice [{}], between time [{} - {}]. Found [{}] items in previous query.",
             logPrefix,
             state.queryCount,
@@ -136,7 +135,7 @@ import org.slf4j.Logger
             .via(deserializeAndAddOffset(state.latest)))
       } else {
         if (log.isDebugEnabled)
-          log.debugN(
+          log.debug(
             "{} query [{}] from slice [{}] completed. Found [{}] items in previous query.",
             logPrefix,
             state.queryCount,
@@ -149,7 +148,7 @@ import org.slf4j.Logger
 
     val currentTimestamp = InstantFactory.now()
     if (log.isDebugEnabled())
-      log.debugN(
+      log.debug(
         "{} query slice [{}], from time [{}] until now [{}].",
         logPrefix,
         slice,
@@ -174,7 +173,7 @@ import org.slf4j.Logger
     val initialOffset = toTimestampOffset(offset)
 
     if (log.isDebugEnabled())
-      log.debugN("Starting {} query from slice [{}], from time [{}].", logPrefix, slice, initialOffset.timestamp)
+      log.debug("Starting {} query from slice [{}], from time [{}].", logPrefix, slice, initialOffset.timestamp)
 
     def nextOffset(state: QueryState, envelope: Envelope): QueryState = {
       val offset = extractOffset(envelope)
@@ -198,7 +197,7 @@ import org.slf4j.Logger
         if (log.isDebugEnabled()) {
           if (state.latestBacktracking.seen.nonEmpty &&
             offset.timestamp.isAfter(state.latestBacktracking.timestamp.plus(firstBacktrackingQueryWindow)))
-            log.debugN(
+            log.debug(
               "{} next offset is outside the backtracking window, latestBacktracking: [{}], offset: [{}]",
               logPrefix,
               state.latestBacktracking,
@@ -221,7 +220,7 @@ import org.slf4j.Logger
 
         if (log.isDebugEnabled)
           delay.foreach { d =>
-            log.debugN(
+            log.debug(
               "{} query [{}] from slice [{}] delay next [{}] ms.",
               logPrefix,
               state.queryCount,
@@ -308,7 +307,7 @@ import org.slf4j.Logger
             " in backtracking mode,"
           else
             ""
-        log.debugN(
+        log.debug(
           "{} next query [{}]{} from slice [{}], between time [{} - {}]. {}",
           logPrefix,
           newState.queryCount,
@@ -351,7 +350,7 @@ import org.slf4j.Logger
               throw new IllegalStateException(
                 s"Too many events stored with the same timestamp [$currentTimestamp], buffer size [${settings.querySettings.bufferSize}]")
             }
-            log.traceN(
+            log.trace(
               "filtering [{}] [{}] as db timestamp is the same as last offset and is in seen [{}]",
               item.persistenceId,
               item.seqNr,

diff --git a/core/src/main/scala/akka/persistence/dynamodb/internal/ContinuousQuery.scala b/core/src/main/scala/akka/persistence/dynamodb/internal/ContinuousQuery.scala
@@ -4,15 +4,15 @@
 
 package akka.persistence.dynamodb.internal
 
-import scala.concurrent.Future
 import scala.concurrent.duration._
+import scala.concurrent.ExecutionContext
+import scala.concurrent.Future
 import scala.util.Failure
 import scala.util.Success
 import scala.util.Try
 
 import akka.NotUsed
 import akka.annotation.InternalApi
-import akka.dispatch.ExecutionContexts
 import akka.stream.Attributes
 import akka.stream.Outlet
 import akka.stream.SourceShape
@@ -118,7 +118,7 @@ final private[dynamodb] class ContinuousQuery[S, T](
             beforeQuery(state) match {
               case None => runNextQuery()
               case Some(beforeQueryFuture) =>
-                beforeQueryFuture.onComplete(beforeQueryCallback.invoke)(ExecutionContexts.parasitic)
+                beforeQueryFuture.onComplete(beforeQueryCallback.invoke)(ExecutionContext.parasitic)
             }
         }
       }

diff --git a/core/src/main/scala/akka/persistence/dynamodb/internal/JournalDao.scala b/core/src/main/scala/akka/persistence/dynamodb/internal/JournalDao.scala
@@ -16,7 +16,6 @@ import scala.jdk.FutureConverters._
 import akka.Done
 import akka.actor.typed.ActorSystem
 import akka.annotation.InternalApi
-import akka.dispatch.ExecutionContexts
 import akka.persistence.Persistence
 import akka.persistence.dynamodb.DynamoDBSettings
 import akka.persistence.typed.PersistenceId
@@ -115,7 +114,7 @@ import software.amazon.awssdk.services.dynamodb.model.Update
             response.consumedCapacity.capacityUnits)
         }
       }
-      result.map(_ => Done)(ExecutionContexts.parasitic)
+      result.map(_ => Done)(ExecutionContext.parasitic)
     } else {
       val writeItems =
         events.map { item =>
@@ -143,10 +142,10 @@ import software.amazon.awssdk.services.dynamodb.model.Update
         }
       }
       result
-        .map(_ => Done)(ExecutionContexts.parasitic)
+        .map(_ => Done)(ExecutionContext.parasitic)
         .recoverWith { case c: CompletionException =>
           Future.failed(c.getCause)
-        }(ExecutionContexts.parasitic)
+        }(ExecutionContext.parasitic)
     }
 
   }
@@ -190,7 +189,7 @@ import software.amazon.awssdk.services.dynamodb.model.Update
     result
       .recoverWith { case c: CompletionException =>
         Future.failed(c.getCause)
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
   }
 
   private def readLowestSequenceNr(persistenceId: String): Future[Long] = {
@@ -220,7 +219,7 @@ import software.amazon.awssdk.services.dynamodb.model.Update
     result
       .recoverWith { case c: CompletionException =>
         Future.failed(c.getCause)
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
   }
 
   def deleteEventsTo(persistenceId: String, toSequenceNr: Long, resetSequenceNumber: Boolean): Future[Unit] = {
@@ -284,10 +283,10 @@ import software.amazon.awssdk.services.dynamodb.model.Update
         }
       }
       result
-        .map(_ => ())(ExecutionContexts.parasitic)
+        .map(_ => ())(ExecutionContext.parasitic)
         .recoverWith { case c: CompletionException =>
           Future.failed(c.getCause)
-        }(ExecutionContexts.parasitic)
+        }(ExecutionContext.parasitic)
     }
 
     // TransactWriteItems has a limit of 100
@@ -317,7 +316,7 @@ import software.amazon.awssdk.services.dynamodb.model.Update
     result
       .recoverWith { case c: CompletionException =>
         Future.failed(c.getCause)
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
   }
 
   def updateEventExpiry(
@@ -376,10 +375,10 @@ import software.amazon.awssdk.services.dynamodb.model.Update
         }
       }
       result
-        .map(_ => ())(ExecutionContexts.parasitic)
+        .map(_ => ())(ExecutionContext.parasitic)
         .recoverWith { case c: CompletionException =>
           Future.failed(c.getCause)
-        }(ExecutionContexts.parasitic)
+        }(ExecutionContext.parasitic)
     }
 
     // TransactWriteItems has a limit of 100
@@ -409,7 +408,7 @@ import software.amazon.awssdk.services.dynamodb.model.Update
     result
       .recoverWith { case c: CompletionException =>
         Future.failed(c.getCause)
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
   }
 
 }
diff --git a/core/src/main/scala/akka/persistence/dynamodb/internal/QueryDao.scala b/core/src/main/scala/akka/persistence/dynamodb/internal/QueryDao.scala
@@ -8,14 +8,14 @@ import java.time.Instant
 import java.util.concurrent.CompletionException
 import java.util.{ Map => JMap }
 
+import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.jdk.CollectionConverters._
 import scala.jdk.FutureConverters._
 
 import akka.NotUsed
 import akka.actor.typed.ActorSystem
 import akka.annotation.InternalApi
-import akka.dispatch.ExecutionContexts
 import akka.persistence.dynamodb.DynamoDBSettings
 import akka.persistence.typed.PersistenceId
 import akka.stream.scaladsl.Source
@@ -303,7 +303,7 @@ import software.amazon.awssdk.services.dynamodb.model.QueryRequest
       }
       .recoverWith { case c: CompletionException =>
         Future.failed(c.getCause)
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
   }
 
   def loadEvent(persistenceId: String, seqNr: Long, includePayload: Boolean): Future[Option[SerializedJournalItem]] = {
@@ -353,7 +353,7 @@ import software.amazon.awssdk.services.dynamodb.model.QueryRequest
       }
       .recoverWith { case c: CompletionException =>
         Future.failed(c.getCause)
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
 
   }
 

diff --git a/core/src/main/scala/akka/persistence/dynamodb/internal/SnapshotDao.scala b/core/src/main/scala/akka/persistence/dynamodb/internal/SnapshotDao.scala
@@ -17,7 +17,6 @@ import scala.jdk.FutureConverters._
 import akka.NotUsed
 import akka.actor.typed.ActorSystem
 import akka.annotation.InternalApi
-import akka.dispatch.ExecutionContexts
 import akka.persistence.Persistence
 import akka.persistence.SnapshotSelectionCriteria
 import akka.persistence.dynamodb.DynamoDBSettings
@@ -106,7 +105,7 @@ import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest
       }
       .recoverWith { case c: CompletionException =>
         Future.failed(c.getCause)
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
   }
 
   private def itemHasExpired(item: JMap[String, AttributeValue]): Boolean = {
@@ -173,10 +172,10 @@ import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest
     }
 
     result
-      .map(_ => ())(ExecutionContexts.parasitic)
+      .map(_ => ())(ExecutionContext.parasitic)
       .recoverWith { case c: CompletionException =>
         Future.failed(c.getCause)
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
   }
 
   def delete(persistenceId: String, criteria: SnapshotSelectionCriteria): Future[Unit] = {
@@ -212,7 +211,7 @@ import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest
     }
 
     result
-      .map(_ => ())(ExecutionContexts.parasitic)
+      .map(_ => ())(ExecutionContext.parasitic)
       // ignore if the criteria conditional check failed
       .recover {
         case _: ConditionalCheckFailedException => ()
@@ -221,7 +220,7 @@ import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest
             case _: ConditionalCheckFailedException => ()
             case cause                              => throw cause
           }
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
   }
 
   def updateExpiry(
@@ -263,7 +262,7 @@ import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest
     }
 
     result
-      .map(_ => ())(ExecutionContexts.parasitic)
+      .map(_ => ())(ExecutionContext.parasitic)
       // ignore if the criteria conditional check failed
       .recover {
         case _: ConditionalCheckFailedException => ()
@@ -272,7 +271,7 @@ import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest
             case _: ConditionalCheckFailedException => ()
             case cause                              => throw cause
           }
-      }(ExecutionContexts.parasitic)
+      }(ExecutionContext.parasitic)
   }
 
   // Used from `BySliceQuery` (only if settings.querySettings.startFromSnapshotEnabled).

diff --git a/core/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournal.scala b/core/src/main/scala/akka/persistence/dynamodb/journal/DynamoDBJournal.scala
@@ -18,7 +18,6 @@ import akka.actor.ActorRef
 import akka.actor.typed.ActorSystem
 import akka.actor.typed.scaladsl.adapter._
 import akka.annotation.InternalApi
-import akka.dispatch.ExecutionContexts
 import akka.event.Logging
 import akka.persistence.AtomicWrite
 import akka.persistence.PersistentRepr
@@ -226,7 +225,7 @@ private[dynamodb] final class DynamoDBJournal(config: Config, cfgPath: String)
       case Some(f) =>
         log.debug("Write in progress for [{}], deferring replayMessages until write completed", persistenceId)
         // we only want to make write - replay sequential, not fail if previous write failed
-        f.recover { case _ => Done }(ExecutionContexts.parasitic)
+        f.recover { case _ => Done }(ExecutionContext.parasitic)
       case None => FutureDone
     }
     pendingWrite.flatMap { _ =>

diff --git a/core/src/main/scala/akka/persistence/dynamodb/query/javadsl/DynamoDBReadJournal.scala b/core/src/main/scala/akka/persistence/dynamodb/query/javadsl/DynamoDBReadJournal.scala
@@ -9,12 +9,12 @@ import java.util
 import java.util.Optional
 import java.util.concurrent.CompletionStage
 
+import scala.concurrent.ExecutionContext
 import scala.jdk.CollectionConverters._
 import scala.jdk.FutureConverters._
 import scala.jdk.OptionConverters.RichOption
 
 import akka.NotUsed
-import akka.dispatch.ExecutionContexts
 import akka.japi.Pair
 import akka.persistence.dynamodb.query.scaladsl
 import akka.persistence.query.Offset
@@ -136,7 +136,7 @@ final class DynamoDBReadJournal(delegate: scaladsl.DynamoDBReadJournal)
     delegate.eventsBySlicesStartingFromSnapshots(entityType, minSlice, maxSlice, offset, transformSnapshot(_)).asJava
 
   override def timestampOf(persistenceId: String, sequenceNr: Long): CompletionStage[Optional[Instant]] =
-    delegate.timestampOf(persistenceId, sequenceNr).map(_.toJava)(ExecutionContexts.parasitic).asJava
+    delegate.timestampOf(persistenceId, sequenceNr).map(_.toJava)(ExecutionContext.parasitic).asJava
 
   override def loadEnvelope[Event](persistenceId: String, sequenceNr: Long): CompletionStage[EventEnvelope[Event]] =
     delegate.loadEnvelope[Event](persistenceId, sequenceNr).asJava