This is an automated email from the ASF dual-hosted git repository.
pjfanning 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 8181d87721 test: fix flaky FusingSpec activeStage assertion (#3496)
8181d87721 is described below
commit 8181d877216f8135d00c3eca6a2f1e9ad0d861df
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Aug 31 17:51:43 2026 +0100
test: fix flaky FusingSpec activeStage assertion (#3496)
Motivation:
FusingSpec "leave activeStage cleared when a stage actor callback completes
its
stage" (added in #3461) intermittently failed observing the downstream
IgnoreSink
as activeStage. The stage the test guards against leaking - the async
source - was
cleared correctly; the sink was still being finalized. The completion
future the
test awaits (Sink.ignore's materialized value) is completed from within the
sink's
own onUpstreamFinish, while activeStage still references the sink, before
the
interpreter runs releaseStage to clear it. The test thread then read the
non-volatile activeStage in a race with that clear.
Modification:
Poll the final assertion with awaitAssert so it settles once the
interpreter goes
idle, instead of reading activeStage once and racing the clear. Rename the
local
to avoid shadowing GraphStageLogic.interpreter in the inner stage.
Result:
The test asserts the same invariant without the timing race.
Tests:
- sbt "stream-tests/testOnly org.apache.pekko.stream.FusingSpec" - 24 passed
References:
Fixes #3494
---
.../src/test/scala/org/apache/pekko/stream/FusingSpec.scala | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
index e5ada5fedf..d54d09cd5b 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
@@ -377,7 +377,11 @@ class FusingSpec extends StreamSpec {
Await.result(ready.future, 3.seconds)
Await.result(stageActorFuture, 3.seconds) ! "complete"
Await.result(done, 3.seconds) should ===(Done)
- Await.result(interpreterPromise.future, 3.seconds).activeStage should
be(null)
+ val completedInterpreter = Await.result(interpreterPromise.future,
3.seconds)
+ // `done` completes from within the downstream sink's own processing, so
the interpreter
+ // may still be finalizing that stage (and briefly hold it as
`activeStage`) when this
+ // thread wakes. Poll until it has settled rather than reading once and
racing the clear.
+ awaitAssert(completedInterpreter.activeStage should be(null))
}
"stop a lazy stage actor dispatch after its handler fails" in {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
