fix: DynamoDB replay rejected events

akka-projection-dynamodb-integration/src/test/scala/akka/projection/dynamodb/TestSourceProviderWithInput.scala

@@ -23,15 +23,19 @@ import akka.persistence.query.typed.EventEnvelope
 import akka.persistence.query.typed.scaladsl.EventTimestampQuery
 import akka.persistence.query.typed.scaladsl.LoadEventQuery
 import akka.projection.BySlicesSourceProvider
+import akka.projection.eventsourced.scaladsl.EventSourcedProvider.LoadEventsByPersistenceIdSourceProvider
 import akka.projection.scaladsl.SourceProvider
 import akka.stream.OverflowStrategy
 import akka.stream.scaladsl.Source
 
-class TestSourceProviderWithInput()(implicit val system: ActorSystem[_])
+class TestSourceProviderWithInput(enableCurrentEventsByPersistenceId: Boolean)(implicit val system: ActorSystem[_])
     extends SourceProvider[TimestampOffset, EventEnvelope[String]]
     with BySlicesSourceProvider
     with EventTimestampQuery
-    with LoadEventQuery {
+    with LoadEventQuery
+    with LoadEventsByPersistenceIdSourceProvider[String] {
+
+  def this()(implicit system: ActorSystem[_]) = this(enableCurrentEventsByPersistenceId = false)
 
   private implicit val ec: ExecutionContext = system.executionContext
   private val persistenceExt = Persistence(system)
@@ -96,4 +100,22 @@ class TestSourceProviderWithInput()(implicit val system: ActorSystem[_])
             s"Event with persistenceId [$persistenceId] and sequenceNr [$sequenceNr] not found."))
     }
   }
+
+  override private[akka] def currentEventsByPersistenceId(
+      persistenceId: String,
+      fromSequenceNr: Long,
+      toSequenceNr: Long): Option[Source[EventEnvelope[String], NotUsed]] = {
+    if (enableCurrentEventsByPersistenceId)
+      Some(
+        Source(
+          envelopes
+            .iterator()
+            .asScala
+            .filter { env =>
+              env.persistenceId == persistenceId && env.sequenceNr >= fromSequenceNr && env.sequenceNr <= toSequenceNr
+            }
+            .toVector))
+    else
+      None
+  }
 }

akka-projection-dynamodb/src/main/mima-filters/1.6.3.backwards.excludes/offset-store.excludes

@@ -0,0 +1,2 @@
+# internal
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.projection.dynamodb.internal.DynamoDBOffsetStore#State.evict")

akka-projection-dynamodb/src/main/resources/reference.conf

@@ -44,6 +44,9 @@ akka.projection.dynamodb {
     }
   }
 
+  # Replay missed events for a particular persistence id when a sequence number is rejected by validation.
+  replay-on-rejected-sequence-numbers = on
+
   # By default it shares DynamoDB client with akka-persistence-dynamodb (write side).
   # To use a separate client for projections this can be
   # set to another config path that defines the config based on

akka-projection-dynamodb/src/main/scala/akka/projection/dynamodb/DynamoDBProjectionSettings.scala

@@ -48,7 +48,8 @@ object DynamoDBProjectionSettings {
       offsetBatchSize = config.getInt("offset-store.offset-batch-size"),
       offsetSliceReadParallelism = config.getInt("offset-store.offset-slice-read-parallelism"),
       timeToLiveSettings = TimeToLiveSettings(config.getConfig("time-to-live")),
-      retrySettings = RetrySettings(config.getConfig("offset-store.retries")))
+      retrySettings = RetrySettings(config.getConfig("offset-store.retries")),
+      replayOnRejectedSequenceNumbers = config.getBoolean("replay-on-rejected-sequence-numbers"))
   }
 
   /**
@@ -72,7 +73,8 @@ final class DynamoDBProjectionSettings private (
     val offsetBatchSize: Int,
     val offsetSliceReadParallelism: Int,
     val timeToLiveSettings: TimeToLiveSettings,
-    val retrySettings: RetrySettings) {
+    val retrySettings: RetrySettings,
+    val replayOnRejectedSequenceNumbers: Boolean) {
 
   // 25 is a hard limit of batch writes in DynamoDB
   require(offsetBatchSize <= 25, s"offset-batch-size must be <= 25, was [$offsetBatchSize]")
@@ -122,6 +124,9 @@ final class DynamoDBProjectionSettings private (
   def withRetrySettings(retrySettings: RetrySettings): DynamoDBProjectionSettings =
     copy(retrySettings = retrySettings)
 
+  def withReplayOnRejectedSequenceNumbers(replayOnRejectedSequenceNumbers: Boolean): DynamoDBProjectionSettings =
+    copy(replayOnRejectedSequenceNumbers = replayOnRejectedSequenceNumbers)
+
   @nowarn("msg=deprecated")
   private def copy(
       timestampOffsetTable: String = timestampOffsetTable,
@@ -132,7 +137,8 @@ final class DynamoDBProjectionSettings private (
       offsetBatchSize: Int = offsetBatchSize,
       offsetSliceReadParallelism: Int = offsetSliceReadParallelism,
       timeToLiveSettings: TimeToLiveSettings = timeToLiveSettings,
-      retrySettings: RetrySettings = retrySettings) =
+      retrySettings: RetrySettings = retrySettings,
+      replayOnRejectedSequenceNumbers: Boolean = replayOnRejectedSequenceNumbers) =
     new DynamoDBProjectionSettings(
       timestampOffsetTable,
       useClient,
@@ -144,10 +150,24 @@ final class DynamoDBProjectionSettings private (
       offsetBatchSize,
       offsetSliceReadParallelism,
       timeToLiveSettings,
-      retrySettings)
+      retrySettings,
+      replayOnRejectedSequenceNumbers)
 
-  override def toString =
-    s"DynamoDBProjectionSettings($timestampOffsetTable, $useClient, $timeWindow, $backtrackingWindow, $warnAboutFilteredEventsInFlow, $offsetBatchSize)"
+  @nowarn("msg=deprecated")
+  override def toString: String =
+    s"DynamoDBProjectionSettings(" +
+    s"timestampOffsetTable=$timestampOffsetTable, " +
+    s"useClient=$useClient, " +
+    s"timeWindow=$timeWindow, " +
+    s"backtrackingWindow=$backtrackingWindow, " +
+    s"keepNumberOfEntries=$keepNumberOfEntries, " +
+    s"evictInterval=$evictInterval, " +
+    s"warnAboutFilteredEventsInFlow=$warnAboutFilteredEventsInFlow, " +
+    s"offsetBatchSize=$offsetBatchSize, " +
+    s"offsetSliceReadParallelism=$offsetSliceReadParallelism, " +
+    s"timeToLiveSettings=$timeToLiveSettings, " +
+    s"retrySettings=$retrySettings, " +
+    s"replayOnRejectedSequenceNumbers=$replayOnRejectedSequenceNumbers)"
 }
 
 object TimeToLiveSettings {

akka-projection-dynamodb/src/main/scala/akka/projection/dynamodb/internal/DynamoDBOffsetStore.scala

@@ -203,7 +203,7 @@ private[projection] class DynamoDBOffsetStore(
     projectionId: ProjectionId,
     sourceProvider: Option[BySlicesSourceProvider],
     system: ActorSystem[_],
-    settings: DynamoDBProjectionSettings,
+    val settings: DynamoDBProjectionSettings,
     client: DynamoDbAsyncClient,
     clock: Clock = Clock.systemUTC()) {
 
@@ -217,7 +217,7 @@ private[projection] class DynamoDBOffsetStore(
   }
 
   private val logger = LoggerFactory.getLogger(this.getClass)
-  private val logPrefix = s"${projectionId.name} [$minSlice-$maxSlice]:"
+  val logPrefix = s"${projectionId.name} [$minSlice-$maxSlice]:"
 
   private val dao = new OffsetStoreDao(system, settings, projectionId, client)
 
@@ -605,10 +605,14 @@ private[projection] class DynamoDBOffsetStore(
             FutureAccepted
           } else if (!recordWithOffset.fromBacktracking) {
             logUnexpected()
+            // Rejected will trigger replay of missed events, if replay-on-rejected-sequence-numbers is enabled
+            // and SourceProvider supports it.
             FutureRejectedSeqNr
           } else {
             logUnexpected()
-            // This will result in projection restart (with normal configuration)
+            // Rejected will trigger replay of missed events, if replay-on-rejected-sequence-numbers is enabled
+            // and SourceProvider supports it.
+            // Otherwise this will result in projection restart (with normal configuration).
             FutureRejectedBacktrackingSeqNr
           }
         } else if (seqNr == 1) {
@@ -682,6 +686,8 @@ private[projection] class DynamoDBOffsetStore(
             previousTimestamp,
             currentState.startTimestampBySlice(slice),
             settings.backtrackingWindow)
+          // Rejected will trigger replay of missed events, if replay-on-rejected-sequence-numbers is enabled
+          // and SourceProvider supports it.
           RejectedBacktrackingSeqNr
         } else {
           // This may happen rather frequently when using `publish-events`, after reconnecting and such.
@@ -690,6 +696,8 @@ private[projection] class DynamoDBOffsetStore(
             seqNr,
             pid,
             recordWithOffset.offset)
+          // Rejected will trigger replay of missed events, if replay-on-rejected-sequence-numbers is enabled
+          // and SourceProvider supports it.
           // Backtracking will emit missed event again.
           RejectedSeqNr
         }