Fix tests by making them respect sequence_number order

Author: tiagomota

core/src/test/scala/akka/persistence/postgres/journal/PostgresJournalSpec.scala

@@ -62,7 +62,6 @@ trait PartitionedJournalSpecTestCases {
   "A journal" must {
     "store events concurrently without any gaps or duplicates among ordering (offset) values" in {
       //given
-      val perId = "perId-1"
       val numOfSenders = 5
       val batchSize = 1000
       val senders = List.fill(numOfSenders)(TestProbe()).zipWithIndex
@@ -72,29 +71,31 @@ trait PartitionedJournalSpecTestCases {
         .sequence {
           senders.map { case (sender, idx) =>
             Future {
-              writeMessages((idx * batchSize) + 1, (idx + 1) * batchSize, perId, sender.ref, writerUuid)
+              writeMessages((idx * batchSize) + 1, (idx + 1) * batchSize, s"perId-${idx + 1}", sender.ref, writerUuid)
             }
           }
         }
         .futureValue(Timeout(Span(1, Minute)))
 
       //then
-      val journalOps = new ScalaPostgresReadJournalOperations(system)
-      journalOps.withCurrentEventsByPersistenceId()(perId) { tp =>
-        tp.request(Long.MaxValue)
-        val replayedMessages = (1 to batchSize * numOfSenders).map { _ =>
-          tp.expectNext()
-        }
-        tp.expectComplete()
-        val orderings = replayedMessages.map(_.offset).collect { case Sequence(value) =>
-          value
-        }
-        orderings.size should equal(batchSize * numOfSenders)
-        val minOrd = orderings.min
-        val maxOrd = orderings.max
-        val expectedOrderings = (minOrd to maxOrd).toList
+      senders.foreach { case (_, idx) =>
+        val journalOps = new ScalaPostgresReadJournalOperations(system)
+        journalOps.withCurrentEventsByPersistenceId()(s"perId-${idx + 1}") { tp =>
+          tp.request(Long.MaxValue)
+          val replayedMessages = (1 to batchSize).map { _ =>
+            tp.expectNext()
+          }
+          tp.expectComplete()
+          val orderings = replayedMessages.map(_.offset).collect { case Sequence(value) =>
+            value
+          }
+          orderings.size should equal(batchSize)
+          val minOrd = orderings.min
+          val maxOrd = orderings.max
+          val expectedOrderings = (minOrd to maxOrd).toList
 
-        (orderings.sorted should contain).theSameElementsInOrderAs(expectedOrderings)
+          (orderings.sorted should contain).theSameElementsInOrderAs(expectedOrderings)
+        }
       }
     }
   }

core/src/test/scala/akka/persistence/postgres/query/JournalSequenceActorTest.scala

@@ -6,7 +6,6 @@
 package akka.persistence.postgres.query
 
 import java.util.concurrent.atomic.AtomicLong
-
 import akka.actor.{ ActorRef, ActorSystem }
 import akka.pattern.ask
 import akka.persistence.postgres.config.JournalSequenceRetrievalConfig
@@ -27,6 +26,7 @@ import slick.jdbc.{ JdbcBackend, JdbcCapabilities }
 
 import scala.concurrent.Future
 import scala.concurrent.duration._
+import scala.util.Random
 
 abstract class JournalSequenceActorTest(val schemaType: SchemaType) extends QueryTestSpec(schemaType.configName) {
   private val log = LoggerFactory.getLogger(classOf[JournalSequenceActorTest])
@@ -76,7 +76,7 @@ abstract class JournalSequenceActorTest(val schemaType: SchemaType) extends Quer
               JournalRow(id, deleted = false, "id", id, Array(0.toByte), Nil, emptyJson)
             }
             .grouped(10000)
-            .mapAsync(4) { rows =>
+            .mapAsync(1) { rows =>
               db.run(journalTable.forceInsertAll(rows))
             }
             .runWith(Sink.ignore)
@@ -112,7 +112,7 @@ abstract class JournalSequenceActorTest(val schemaType: SchemaType) extends Quer
               JournalRow(id, deleted = false, "id", id, Array(0.toByte), Nil, emptyJson)
             }
             .grouped(10000)
-            .mapAsync(4) { rows =>
+            .mapAsync(1) { rows =>
               db.run(journalTable.forceInsertAll(rows))
             }
             .runWith(Sink.ignore)
@@ -145,7 +145,7 @@ abstract class JournalSequenceActorTest(val schemaType: SchemaType) extends Quer
               JournalRow(id, deleted = false, "id", id, Array(0.toByte), Nil, emptyJson)
             }
             .grouped(10000)
-            .mapAsync(4) { rows =>
+            .mapAsync(1) { rows =>
               db.run(journalTable.forceInsertAll(rows))
             }
             .runWith(Sink.ignore)