log warning every 1000

akka · Nov 29, 2024 · 2f237b9 · 2f237b9
1 parent 673e606
commit 2f237b9
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 19 deletions.
diff --git a/...ction-dynamodb/src/main/scala/akka/projection/dynamodb/internal/DynamoDBOffsetStore.scala b/...ction-dynamodb/src/main/scala/akka/projection/dynamodb/internal/DynamoDBOffsetStore.scala
@@ -622,7 +622,6 @@ private[projection] class DynamoDBOffsetStore(
           // always accept starting from snapshots when there was no previous event seen
           FutureAccepted
         } else {
-          println(s"# validateEventTimestamp $recordWithOffset") // FIXME
           validateEventTimestamp(currentState, recordWithOffset)
         }
       } else {

diff --git a/...on-dynamodb/src/main/scala/akka/projection/dynamodb/internal/DynamoDBProjectionImpl.scala b/...on-dynamodb/src/main/scala/akka/projection/dynamodb/internal/DynamoDBProjectionImpl.scala
@@ -87,6 +87,7 @@ private[projection] object DynamoDBProjectionImpl {
   }
 
   private val loadEnvelopeCounter = new AtomicLong
+  private val replayRejectedCounter = new AtomicLong
 
   def loadEnvelope[Envelope](env: Envelope, sourceProvider: SourceProvider[_, Envelope])(
       implicit
@@ -223,12 +224,7 @@ private[projection] object DynamoDBProjectionImpl {
                 offsetStore.storedSeqNr(persistenceId).flatMap { storedSeqNr =>
                   val fromSeqNr = storedSeqNr + 1
                   val toSeqNr = originalEventEnvelope.sequenceNr
-                  log.debug(
-                    s"{} Replaying events after rejected sequence number. PersistenceId [{}], replaying from seqNr [{}] to [{}].",
-                    offsetStore.logPrefix,
-                    persistenceId,
-                    fromSeqNr,
-                    toSeqNr)
+                  logReplayRejected(offsetStore, persistenceId, fromSeqNr, toSeqNr)
                   provider.currentEventsByPersistenceId(persistenceId, fromSeqNr, toSeqNr) match {
                     case Some(querySource) =>
                       querySource
@@ -485,12 +481,7 @@ private[projection] object DynamoDBProjectionImpl {
                 offsetStore.storedSeqNr(persistenceId).flatMap { storedSeqNr =>
                   val fromSeqNr = storedSeqNr + 1
                   val toSeqNr = originalEventEnvelope.sequenceNr
-                  log.debug(
-                    s"{} Replaying events after rejected sequence number. PersistenceId [{}], replaying from seqNr [{}] to [{}].",
-                    offsetStore.logPrefix,
-                    persistenceId,
-                    fromSeqNr,
-                    toSeqNr)
+                  logReplayRejected(offsetStore, persistenceId, fromSeqNr, toSeqNr)
                   provider.currentEventsByPersistenceId(persistenceId, fromSeqNr, toSeqNr) match {
                     case Some(querySource) =>
                       querySource
@@ -585,12 +576,7 @@ private[projection] object DynamoDBProjectionImpl {
               offsetStore.storedSeqNr(persistenceId).flatMap { storedSeqNr =>
                 val fromSeqNr = storedSeqNr + 1
                 val toSeqNr = originalEventEnvelope.sequenceNr
-                log.debug(
-                  s"{} Replaying events after rejected sequence number. PersistenceId [{}], replaying from seqNr [{}] to [{}].",
-                  offsetStore.logPrefix,
-                  persistenceId,
-                  fromSeqNr,
-                  toSeqNr)
+                logReplayRejected(offsetStore, persistenceId, fromSeqNr, toSeqNr)
                 provider.currentEventsByPersistenceId(persistenceId, fromSeqNr, toSeqNr) match {
                   case Some(querySource) =>
                     querySource
@@ -699,6 +685,20 @@ private[projection] object DynamoDBProjectionImpl {
       .via(handler)
   }
 
+  private def logReplayRejected(
+      offsetStore: DynamoDBOffsetStore,
+      persistenceId: String,
+      fromSeqNr: Long,
+      toSeqNr: Long): Unit = {
+    val msg =
+      "{} Replaying events after rejected sequence number. PersistenceId [{}], replaying from seqNr [{}] to [{}]. Replay count [{}]."
+    val c = replayRejectedCounter.incrementAndGet()
+    if (c == 1 || c % 1000 == 0)
+      log.warn(msg, offsetStore.logPrefix, persistenceId, fromSeqNr, toSeqNr, c)
+    else
+      log.debug(msg, offsetStore.logPrefix, persistenceId, fromSeqNr, toSeqNr, c)
+  }
+
   private def triggerReplayIfPossible[Offset, Envelope](
       sourceProvider: SourceProvider[Offset, Envelope],
       offsetStore: DynamoDBOffsetStore,