limit windowing support snapshoted events

core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseJournalDaoWithReadMessages.scala

@@ -19,7 +19,7 @@ import pekko.NotUsed
 import pekko.actor.Scheduler
 import pekko.annotation.InternalApi
 import pekko.persistence.PersistentRepr
-import pekko.persistence.jdbc.journal.dao.FlowControl.{ Continue, ContinueDelayed, Stop }
+import pekko.persistence.jdbc.journal.dao.FlowControl.{ Continue, ContinueDelayed, Fallback, Stop }
 import pekko.stream.Materializer
 import pekko.stream.scaladsl.{ Sink, Source }
 
@@ -51,17 +51,24 @@ trait BaseJournalDaoWithReadMessages extends JournalDaoWithReadMessages {
       toSequenceNr: Long,
       batchSize: Int,
       refreshInterval: Option[(FiniteDuration, Scheduler)]) = {
+    val firstSequenceVer: Long = Math.max(1, fromSequenceNr)
     Source
-      .unfoldAsync[(Long, FlowControl), Seq[Try[(PersistentRepr, Long)]]]((Math.max(1, fromSequenceNr), Continue)) {
+      .unfoldAsync[(Long, FlowControl), Seq[Try[(PersistentRepr, Long)]]]((firstSequenceVer, Continue)) {
         case (from, control) =>
-          def limitWindow(from: Long): Long = {
-            math.min(from + batchSize, toSequenceNr)
+          def limitWindow(from: Long)(implicit fallback: Boolean): Long = {
+            if (fallback || batchSize <= 0 || (Long.MaxValue - batchSize) < from) {
+              toSequenceNr
+            } else {
+              Math.min(from + batchSize, toSequenceNr)
+            }
           }
 
-          def retrieveNextBatch(): Future[Option[((Long, FlowControl), Seq[Try[(PersistentRepr, Long)]])]] = {
+          def retrieveNextBatch(implicit fallback: Boolean = false)
+              : Future[Option[((Long, FlowControl), Seq[Try[(PersistentRepr, Long)]])]] = {
             for {
               xs <- messages(persistenceId, from, limitWindow(from), batchSize).runWith(Sink.seq)
             } yield {
+              val isEmptyEvents = xs.isEmpty
               val hasMoreEvents = xs.size == batchSize
               // Events are ordered by sequence number, therefore the last one is the largest)
               val lastSeqNrInBatch: Option[Long] = xs.lastOption match {
@@ -73,6 +80,7 @@ trait BaseJournalDaoWithReadMessages extends JournalDaoWithReadMessages {
               val nextControl: FlowControl =
                 if (hasLastEvent || from > toSequenceNr) Stop
                 else if (hasMoreEvents) Continue
+                else if (isEmptyEvents && from == firstSequenceVer) Fallback
                 else if (refreshInterval.isEmpty) Stop
                 else ContinueDelayed
 
@@ -91,6 +99,7 @@ trait BaseJournalDaoWithReadMessages extends JournalDaoWithReadMessages {
             case ContinueDelayed =>
               val (delay, scheduler) = refreshInterval.get
               pekko.pattern.after(delay, scheduler)(retrieveNextBatch())
+            case Fallback => retrieveNextBatch(fallback = true)
           }
       }
   }

core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/FlowControl.scala

@@ -27,6 +27,11 @@ private[jdbc] object FlowControl {
    */
   case object ContinueDelayed extends FlowControl
 
+  /**
+   * if the limited windows unable query anything, then fallback to full windows.
+   */
+  case object Fallback extends FlowControl
+
   /** Stop querying - used when we reach the desired offset */
   case object Stop extends FlowControl
 }

integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/integration/LimitWindowingStreamTest.scala

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.pekko.persistence.jdbc.integration
+
+import org.apache.pekko.persistence.jdbc.journal.dao.LimitWindowingStreamTest
+import org.apache.pekko.persistence.jdbc.query.{ MysqlCleaner, OracleCleaner, PostgresCleaner, SqlServerCleaner }
+
+class PostgresLimitWindowingStreamTest
+    extends LimitWindowingStreamTest("postgres-application.conf")
+    with PostgresCleaner
+
+class MySQLLimitWindowingStreamTest
+    extends LimitWindowingStreamTest("mysql-application.conf")
+    with MysqlCleaner
+
+class OracleLimitWindowingStreamTest
+    extends LimitWindowingStreamTest("oracle-application.conf")
+    with OracleCleaner
+
+class SqlServerLimitWindowingStreamTest
+    extends LimitWindowingStreamTest("sqlserver-application.conf")
+    with SqlServerCleaner