This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 97a03a3d2b fix: detect delayed messages in
ManualTime.expectNoMessageFor (#3329)
97a03a3d2b is described below
commit 97a03a3d2b8e49c7d94cef574511cf1f39923327
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Mon Jul 13 20:41:25 2026 +0800
fix: detect delayed messages in ManualTime.expectNoMessageFor (#3329)
Motivation:
ManualTime.expectNoMessageFor used a zero-timeout probe poll after
advancing manual time, allowing asynchronously dispatched messages to be missed.
Modification:
Use the configured no-message timeout in both Scala and Java DSLs, document
the behavior, and add deterministic regression coverage for delayed delivery.
Result:
ManualTime.expectNoMessageFor now fails when a message triggered during the
time advance reaches a probe asynchronously.
Tests:
- sbt "+actor-testkit-typed / Test / testOnly
org.apache.pekko.actor.testkit.typed.scaladsl.ManualTimeSpec" (passed on Scala
2.13.18 and 3.3.8)
- sbt "actor-testkit-typed / Test / test" (175 total, 174 passed, 1 ignored)
- sbt "+headerCheckAll" (passed)
- sbt checkCodeStyle (passed)
- native scalafmt diff and new-file checks (passed)
- Qoder stdout review (no must-fix findings)
- sbt sortImports (environment failure: Scalafix/Scalameta
NoSuchMethodError)
- sbt validatePullRequest (interrupted at user request)
References:
Fixes #3235
---
.../actor/testkit/typed/javadsl/ManualTime.scala | 8 +-
.../actor/testkit/typed/scaladsl/ManualTime.scala | 10 +-
.../testkit/typed/scaladsl/ManualTimeSpec.scala | 122 +++++++++++++++++++++
3 files changed, 137 insertions(+), 3 deletions(-)
diff --git
a/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/javadsl/ManualTime.scala
b/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/javadsl/ManualTime.scala
index abb91dfa6a..9f0d7c35c5 100644
---
a/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/javadsl/ManualTime.scala
+++
b/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/javadsl/ManualTime.scala
@@ -72,10 +72,16 @@ final class ManualTime(delegate:
pekko.testkit.ExplicitlyTriggeredScheduler) {
*/
def timePasses(amount: Duration): Unit = delegate.timePasses(amount.toScala)
+ /**
+ * Advance the clock by the specified duration and assert that the probes
receive no messages.
+ *
+ * After advancing the clock, each probe waits for the configured
+ * `pekko.actor.testkit.typed.expect-no-message-default` duration so that
asynchronously dispatched messages can arrive.
+ */
@varargs
def expectNoMessageFor(duration: Duration, on: TestProbe[?]*): Unit = {
delegate.timePasses(duration.toScala)
- on.foreach(_.expectNoMessage(Duration.ZERO))
+ on.foreach(_.expectNoMessage())
}
}
diff --git
a/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTime.scala
b/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTime.scala
index ae765bcbfe..c631307343 100644
---
a/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTime.scala
+++
b/actor-testkit-typed/src/main/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTime.scala
@@ -14,7 +14,7 @@
package org.apache.pekko.actor.testkit.typed.scaladsl
import scala.annotation.varargs
-import scala.concurrent.duration.{ Duration, FiniteDuration }
+import scala.concurrent.duration.FiniteDuration
import org.apache.pekko
import pekko.actor.typed.ActorSystem
@@ -72,10 +72,16 @@ final class ManualTime(delegate:
pekko.testkit.ExplicitlyTriggeredScheduler) {
*/
def timePasses(amount: FiniteDuration): Unit = delegate.timePasses(amount)
+ /**
+ * Advance the clock by the specified duration and assert that the probes
receive no messages.
+ *
+ * After advancing the clock, each probe waits for the configured
+ * `pekko.actor.testkit.typed.expect-no-message-default` duration so that
asynchronously dispatched messages can arrive.
+ */
@varargs
def expectNoMessageFor(duration: FiniteDuration, on: TestProbe[?]*): Unit = {
delegate.timePasses(duration)
- on.foreach(_.expectNoMessage(Duration.Zero))
+ on.foreach(_.expectNoMessage())
}
}
diff --git
a/actor-testkit-typed/src/test/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTimeSpec.scala
b/actor-testkit-typed/src/test/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTimeSpec.scala
new file mode 100644
index 0000000000..e9e308bd0f
--- /dev/null
+++
b/actor-testkit-typed/src/test/scala/org/apache/pekko/actor/testkit/typed/scaladsl/ManualTimeSpec.scala
@@ -0,0 +1,122 @@
+/*
+ * 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.actor.testkit.typed.scaladsl
+
+import java.time.{ Duration => JDuration }
+import java.util.concurrent.{ CountDownLatch, TimeUnit }
+
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.testkit.typed.scaladsl.ManualTimeSpec.config
+import pekko.actor.typed.DispatcherSelector
+import pekko.actor.typed.scaladsl.Behaviors
+
+import org.scalatest.wordspec.AnyWordSpecLike
+
+import com.typesafe.config.ConfigFactory
+
+object ManualTimeSpec {
+ val config = ConfigFactory
+ .parseString("""
+ manual-time-test-dispatcher {
+ type = Dispatcher
+ executor = "thread-pool-executor"
+ thread-pool-executor.fixed-pool-size = 1
+ throughput = 1
+ }
+ pekko.actor.testkit.typed.expect-no-message-default = 3 s
+ """)
+ .withFallback(ManualTime.config)
+}
+
+class ManualTimeSpec extends ScalaTestWithActorTestKit(config) with
AnyWordSpecLike with LogCapturing {
+
+ private val manualTime = ManualTime()
+
+ private def verifyDelayedMessageIsDetected(expectNoMessageFor:
TestProbe[String] => Unit): Unit = {
+ val target = TestProbe[String]()
+ val timerStarted = TestProbe[Done]()
+ val dispatcherSelector =
DispatcherSelector.fromConfig("manual-time-test-dispatcher")
+
+ spawn(
+ Behaviors.withTimers[String] { timers =>
+ timers.startSingleTimer("unexpected", 1.milli)
+ timerStarted.ref ! Done
+ Behaviors.receiveMessage { message =>
+ target.ref ! message
+ Behaviors.same
+ }
+ },
+ dispatcherSelector)
+ timerStarted.expectMessage(Done)
+
+ val dispatcherBlocked = new CountDownLatch(1)
+ val releaseDispatcher = new CountDownLatch(1)
+ val blocker = spawn(
+ Behaviors.receiveMessage[Done] { _ =>
+ dispatcherBlocked.countDown()
+ releaseDispatcher.await()
+ Behaviors.stopped
+ },
+ dispatcherSelector)
+ blocker ! Done
+ dispatcherBlocked.await(3, TimeUnit.SECONDS) shouldBe true
+
+ val testThread = Thread.currentThread()
+ val watcherDone = new CountDownLatch(1)
+ system.scheduler.scheduleOnce(
+ 2.millis,
+ () => {
+ val watcher = new Thread(
+ () => {
+ try {
+ while (releaseDispatcher.getCount != 0 && testThread.getState !=
Thread.State.TIMED_WAITING)
+ Thread.onSpinWait()
+ if (testThread.getState == Thread.State.TIMED_WAITING)
+ releaseDispatcher.countDown()
+ } finally watcherDone.countDown()
+ },
+ "manual-time-expect-no-message-watcher")
+ watcher.setDaemon(true)
+ watcher.start()
+ })(system.executionContext)
+
+ try {
+ intercept[AssertionError] {
+ expectNoMessageFor(target)
+ }.getMessage should include("Received unexpected message unexpected")
+ } finally {
+ releaseDispatcher.countDown()
+ watcherDone.await(3, TimeUnit.SECONDS) shouldBe true
+ }
+ }
+
+ "ManualTime.expectNoMessageFor" must {
+ "detect a delayed message with the Scala DSL" in {
+ verifyDelayedMessageIsDetected(probe =>
manualTime.expectNoMessageFor(2.millis, probe))
+ }
+
+ "detect a delayed message with the Java DSL" in {
+ val javaManualTime =
pekko.actor.testkit.typed.javadsl.ManualTime.get(system)
+ verifyDelayedMessageIsDetected(probe =>
+ javaManualTime.expectNoMessageFor(JDuration.ofMillis(2), probe.asJava))
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
