You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by nv...@apache.org on 2023/08/22 09:19:40 UTC
[incubator-pekko-persistence-cassandra] branch fix-tests created (now e554f2c)
This is an automated email from the ASF dual-hosted git repository.
nvollmar pushed a change to branch fix-tests
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-persistence-cassandra.git
at e554f2c wip
This branch includes the following new commits:
new 3a69923 Fixing flaky test
new e554f2c wip
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-persistence-cassandra] 02/02: wip
Posted by nv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nvollmar pushed a commit to branch fix-tests
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-persistence-cassandra.git
commit e554f2c1d49ce5e9038ba84d4ce14904d22b4433
Author: Nicolas Vollmar <nv...@scaling.ch>
AuthorDate: Tue Aug 22 11:19:34 2023 +0200
wip
---
.github/workflows/unit-tests.yml | 1 +
1 file changed, 1 insertion(+)
diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml
index b4fe285..1def731 100644
--- a/.github/workflows/unit-tests.yml
+++ b/.github/workflows/unit-tests.yml
@@ -6,6 +6,7 @@ on:
branches:
- main
- release-*
+ - fix-tests
tags-ignore: [ v.* ]
schedule:
- cron: '0 2 * * *' # every day 2am
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
[incubator-pekko-persistence-cassandra] 01/02: Fixing flaky test
Posted by nv...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nvollmar pushed a commit to branch fix-tests
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-persistence-cassandra.git
commit 3a699236b3f2406370ad1d97caec454e6d188ce9
Author: Nicolas Vollmar <nv...@scaling.ch>
AuthorDate: Tue Aug 22 11:18:57 2023 +0200
Fixing flaky test
---
.../persistence/cassandra/CassandraLifecycle.scala | 31 ++++++++--------
.../cassandra/query/EventsByTagSpec.scala | 42 +++++++++++++---------
2 files changed, 41 insertions(+), 32 deletions(-)
diff --git a/core/src/test/scala/org/apache/pekko/persistence/cassandra/CassandraLifecycle.scala b/core/src/test/scala/org/apache/pekko/persistence/cassandra/CassandraLifecycle.scala
index a1c3877..89c26a6 100644
--- a/core/src/test/scala/org/apache/pekko/persistence/cassandra/CassandraLifecycle.scala
+++ b/core/src/test/scala/org/apache/pekko/persistence/cassandra/CassandraLifecycle.scala
@@ -17,19 +17,18 @@ import java.time.LocalDateTime
import java.time.ZoneOffset
import java.time.format.DateTimeFormatter
import java.util.concurrent.TimeUnit
-
import scala.concurrent.Await
-
import org.apache.pekko
-import pekko.actor.{ ActorSystem, PoisonPill, Props }
+import pekko.actor.{ActorSystem, PoisonPill, Props}
import pekko.persistence.PersistentActor
-import pekko.testkit.{ TestKitBase, TestProbe }
+import pekko.testkit.{TestKitBase, TestProbe}
import com.datastax.oss.driver.api.core.CqlSession
import com.typesafe.config.ConfigFactory
import org.scalatest._
-import scala.concurrent.duration._
-import scala.util.{ Failure, Success, Try }
+import org.scalatest.concurrent.Eventually.eventually
+import scala.concurrent.duration._
+import scala.util.{Failure, Success, Try}
import pekko.stream.connectors.cassandra.scaladsl.CassandraSession
import pekko.stream.connectors.cassandra.scaladsl.CassandraSessionRegistry
@@ -77,15 +76,17 @@ object CassandraLifecycle {
system.actorOf(
Props(classOf[AwaitPersistenceInit], "persistenceInit" + n, journalPluginId, snapshotPluginId),
"persistenceInit" + n)
- a.tell("hello", probe.ref)
- try {
- probe.expectMsg(5.seconds, "hello")
- } catch {
- case t: Throwable =>
- probe.watch(a)
- a ! PoisonPill
- probe.expectTerminated(a, 10.seconds)
- throw t
+ eventually {
+ a.tell("hello", probe.ref)
+ try {
+ probe.expectMsg(5.seconds, "hello")
+ } catch {
+ case t: Throwable =>
+ probe.watch(a)
+ a ! PoisonPill
+ probe.expectTerminated(a, 10.seconds)
+ throw t
+ }
}
system.log.debug(
"awaitPersistenceInit took {} ms {}",
diff --git a/core/src/test/scala/org/apache/pekko/persistence/cassandra/query/EventsByTagSpec.scala b/core/src/test/scala/org/apache/pekko/persistence/cassandra/query/EventsByTagSpec.scala
index e4e4ca8..ac7688c 100644
--- a/core/src/test/scala/org/apache/pekko/persistence/cassandra/query/EventsByTagSpec.scala
+++ b/core/src/test/scala/org/apache/pekko/persistence/cassandra/query/EventsByTagSpec.scala
@@ -36,6 +36,8 @@ import com.datastax.oss.driver.api.core.CqlIdentifier
import com.datastax.oss.driver.api.core.uuid.Uuids
import com.typesafe.config.{ Config, ConfigFactory }
import org.scalatest.BeforeAndAfterEach
+import org.scalatest.concurrent.Eventually.eventually
+import org.scalatest.time.{ Minutes, Span }
import scala.concurrent.duration._
import pekko.persistence.cassandra.PluginSettings
@@ -1370,23 +1372,29 @@ class EventsByTagPersistenceIdCleanupSpec extends AbstractEventsByTagSpec(Events
"PersistenceId cleanup" must {
"drop state and trigger new persistence id lookup periodically" in {
- val t1: LocalDateTime = LocalDateTime.now(ZoneOffset.UTC).minusDays(2)
- val event1 = PersistentRepr(s"cleanup-1", 1, "cleanup")
- writeTaggedEvent(t1, event1, Set("cleanup-tag"), 1, bucketSize)
-
- val query =
- queries.eventsByTag("cleanup-tag", TimeBasedUUID(Uuids.startOf(t1.toInstant(ZoneOffset.UTC).toEpochMilli - 1L)))
- val probe = query.runWith(TestSink.probe[Any])
- probe.request(10)
- probe.expectNextPF { case e @ EventEnvelope(_, "cleanup", 1L, "cleanup-1") => e }
- probe.expectNoMessage(cleanupPeriod + 250.millis)
-
- // the metadata for pid cleanup should have been removed meaning the next event will be delayed
- val event2 = PersistentRepr(s"cleanup-2", 2, "cleanup")
- writeTaggedEvent(event2, Set("cleanup-tag"), 2, bucketSize)
-
- probe.expectNoMessage(newPersistenceIdScan - 100.millis) // we don't now when exactly the next persistence id scan will be
- probe.expectNextPF { case e @ EventEnvelope(_, "cleanup", 2L, "cleanup-2") => e }
+ // this test depends on new persistenceId can not happening at the wrong time...
+ eventually(timeout(Span(1, Minutes))) {
+ val t1: LocalDateTime = LocalDateTime.now(ZoneOffset.UTC).minusDays(2)
+ val event1 = PersistentRepr(s"cleanup-1", 1, "cleanup")
+ writeTaggedEvent(t1, event1, Set("cleanup-tag"), 1, bucketSize)
+
+ val query =
+ queries.eventsByTag("cleanup-tag",
+ TimeBasedUUID(Uuids.startOf(t1.toInstant(ZoneOffset.UTC).toEpochMilli - 1L)))
+ val probe = query.runWith(TestSink.probe[Any])
+ probe.request(10)
+ probe.expectNextPF { case e @ EventEnvelope(_, "cleanup", 1L, "cleanup-1") => e }
+ // wait for persistence id cleanup
+ probe.expectNoMessage(cleanupPeriod + 250.millis)
+
+ // the metadata for pid cleanup should have been removed meaning the next event will be delayed
+ val event2 = PersistentRepr(s"cleanup-2", 2, "cleanup")
+ writeTaggedEvent(event2, Set("cleanup-tag"), 2, bucketSize)
+
+ // we don't now when exactly the next persistence id scan will be
+ probe.expectNoMessage(newPersistenceIdScan - 100.millis)
+ probe.expectNextPF { case e @ EventEnvelope(_, "cleanup", 2L, "cleanup-2") => e }
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org