make one of the new persistence-tck journal tests optional

docs/src/test/scala/docs/persistence/PersistencePluginDocSpec.scala

@@ -14,17 +14,17 @@
 package docs.persistence
 
 import scala.collection.immutable
-import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.testkit.TestKit
-import com.typesafe.config._
-import org.scalatest.wordspec.AnyWordSpec
-
 import scala.concurrent.Future
-import scala.util.Try
 import scala.concurrent.duration._
+import scala.util.Try
 
-//#plugin-imports
+import com.typesafe.config._
 import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.testkit.TestKit
+import org.scalatest.wordspec.AnyWordSpec
+
+//#plugin-imports
 import pekko.persistence._
 import pekko.persistence.journal._
 import pekko.persistence.snapshot._
@@ -116,8 +116,8 @@ class PersistencePluginDocSpec extends AnyWordSpec {
 }
 
 object SharedLeveldbPluginDocSpec {
-  import org.apache.pekko.actor._
-  import org.apache.pekko.persistence.journal.leveldb.SharedLeveldbJournal
+  import pekko.actor._
+  import pekko.persistence.journal.leveldb.SharedLeveldbJournal
 
   val config =
     """
@@ -153,9 +153,9 @@ trait SharedLeveldbPluginDocSpec {
   val system: ActorSystem
 
   {
-    import org.apache.pekko.actor._
+    import pekko.actor._
     // #shared-store-creation
-    import org.apache.pekko.persistence.journal.leveldb.SharedLeveldbStore
+    import pekko.persistence.journal.leveldb.SharedLeveldbStore
 
     val store = system.actorOf(Props[SharedLeveldbStore](), "store")
     // #shared-store-creation
@@ -192,7 +192,7 @@ class MySnapshotStore extends SnapshotStore {
 
 object PersistenceTCKDoc {
   object example1 {
-    import org.apache.pekko.persistence.journal.JournalSpec
+    import pekko.persistence.journal.JournalSpec
 
     // #journal-tck-scala
     class MyJournalSpec
@@ -208,7 +208,7 @@ object PersistenceTCKDoc {
     // #journal-tck-scala
   }
   object example2 {
-    import org.apache.pekko.persistence.snapshot.SnapshotStoreSpec
+    import pekko.persistence.snapshot.SnapshotStoreSpec
 
     // #snapshot-store-tck-scala
     class MySnapshotStoreSpec
@@ -223,7 +223,7 @@ object PersistenceTCKDoc {
     // #snapshot-store-tck-scala
   }
   object example2b {
-    import org.apache.pekko.persistence.state.DurableStateStoreSpec
+    import pekko.persistence.state.DurableStateStoreSpec
 
     // #durable-state-tck-scala
     class MyDurableStateStoreSpec
@@ -246,7 +246,7 @@ object PersistenceTCKDoc {
   object example3 {
     import java.io.File
 
-    import org.apache.pekko.persistence.journal.JournalSpec
+    import pekko.persistence.journal.JournalSpec
     import org.iq80.leveldb.util.FileUtils
 
     // #journal-tck-before-after-scala

persistence-tck/src/main/scala/org/apache/pekko/persistence/CapabilityFlags.scala

@@ -67,6 +67,16 @@ trait JournalCapabilityFlags extends CapabilityFlags {
    */
   protected def supportsMetadata: CapabilityFlag
 
+  /**
+   * When `true` enables tests which check if the Journal supports a replay window that
+   * spans a deleted prefix.
+   * <p>
+   *   Relates to https://github.com/apache/pekko-persistence-jdbc/pull/517.
+   *   pekko-persistence-dynamodb doesn't yet fully support this.
+   * </p>
+   */
+  protected def supportsReplayWindowSpanningDeletedPrefix: CapabilityFlag
+
 }
 //#journal-flags
 

persistence-tck/src/main/scala/org/apache/pekko/persistence/japi/journal/JavaJournalPerfSpec.scala

@@ -65,4 +65,6 @@ class JavaJournalPerfSpec(config: Config) extends JournalPerfSpec(config) {
   override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = CapabilityFlag.on()
 
   override protected def supportsSerialization: CapabilityFlag = CapabilityFlag.on()
+
+  override protected def supportsReplayWindowSpanningDeletedPrefix: CapabilityFlag = CapabilityFlag.on()
 }

persistence-tck/src/main/scala/org/apache/pekko/persistence/journal/JournalSpec.scala

@@ -66,6 +66,8 @@ abstract class JournalSpec(config: Config)
 
   override protected def supportsMetadata: CapabilityFlag = false
 
+  override protected def supportsReplayWindowSpanningDeletedPrefix: CapabilityFlag = false
+
   override protected def beforeEach(): Unit = {
     super.beforeEach()
     senderProbe = TestProbe()
@@ -259,16 +261,6 @@ abstract class JournalSpec(config: Config)
       receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
     }
 
-    "replay surviving messages within bounds when the replay window spans a deleted prefix" in {
-      val deleteProbe = TestProbe()
-      journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref)
-      deleteProbe.expectMsg(DeleteMessagesSuccess(3L))
-
-      journal ! ReplayMessages(1, 4, Long.MaxValue, pid, receiverProbe.ref)
-      receiverProbe.expectMsg(replayedMessage(4))
-      receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
-    }
-
     "return only recovery success when the upper bound falls within a deleted prefix" in {
       val deleteProbe = TestProbe()
       journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref)
@@ -409,5 +401,17 @@ abstract class JournalSpec(config: Config)
 
       }
     }
+
+    optional(flag = supportsReplayWindowSpanningDeletedPrefix) {
+      "replay surviving messages within bounds when the replay window spans a deleted prefix" in {
+        val deleteProbe = TestProbe()
+        journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref)
+        deleteProbe.expectMsg(DeleteMessagesSuccess(3L))
+
+        journal ! ReplayMessages(1, 4, Long.MaxValue, pid, receiverProbe.ref)
+        receiverProbe.expectMsg(replayedMessage(4))
+        receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
+      }
+    }
   }
 }

persistence-tck/src/test/scala/org/apache/pekko/persistence/journal/inmem/InmemJournalSpec.scala

@@ -20,4 +20,5 @@ import pekko.persistence.journal.JournalSpec
 
 class InmemJournalSpec extends JournalSpec(config = PersistenceSpec.config("inmem", "InmemJournalSpec")) {
   override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = CapabilityFlag.off()
+  override protected def supportsReplayWindowSpanningDeletedPrefix: CapabilityFlag = CapabilityFlag.on()
 }

persistence-tck/src/test/scala/org/apache/pekko/persistence/journal/leveldb/LeveldbJournalJavaSpec.scala

@@ -32,4 +32,7 @@ class LeveldbJournalJavaSpec
   override def supportsRejectingNonSerializableObjects = true
 
   override def supportsSerialization = true
+
+  override def supportsReplayWindowSpanningDeletedPrefix = true
+
 }

persistence-tck/src/test/scala/org/apache/pekko/persistence/journal/leveldb/LeveldbJournalNativeSpec.scala

@@ -33,4 +33,6 @@ class LeveldbJournalNativeSpec
 
   override def supportsSerialization = true
 
+  override def supportsReplayWindowSpanningDeletedPrefix = true
+
 }

persistence-tck/src/test/scala/org/apache/pekko/persistence/journal/leveldb/LeveldbJournalNoAtomicPersistMultipleEventsSpec.scala

@@ -38,4 +38,6 @@ class LeveldbJournalNoAtomicPersistMultipleEventsSpec
 
   override def supportsSerialization = true
 
+  override def supportsReplayWindowSpanningDeletedPrefix = true
+
 }

persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/scaladsl/PersistenceTestKitJournalCompatSpec.scala

@@ -48,6 +48,7 @@ class PersistenceTestKitJournalCompatSpec extends JournalSpec(config = Persisten
 
   override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = true
   override protected def supportsMetadata: CapabilityFlag = true
+  override protected def supportsReplayWindowSpanningDeletedPrefix: CapabilityFlag = true
 }
 
 class PersistenceTestKitSnapshotStoreCompatSpec