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 e520a7c642 test: cover async boundary failure draining (#3296)
e520a7c642 is described below
commit e520a7c642a2768f1b55da8ba390c7754d963183
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Mon Jul 6 01:57:50 2026 +0800
test: cover async boundary failure draining (#3296)
Motivation:
Async boundary batching should preserve elements already produced by an
upstream stage before that stage fails.
Modification:
Add a focused FusingSpec regression using transformed string elements
upstream of an async boundary, then assert the already emitted elements arrive
before the failure.
Result:
Async boundary failure-draining behavior has a direct regression test.
Tests:
- rtk scalafmt --mode diff-ref=origin/main -- passed
- rtk scalafmt --list --mode diff-ref=origin/main -- passed
- rtk git diff --check -- passed
- rtk sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.FusingSpec -- -z \"deliver transformed elements already
produced\"" -- passed, 1 test
- rtk qodercli --help -- passed, confirmed -p/--print, --output-format,
--cwd, and --attachment
- qodercli stdout review of /tmp/project-review.diff -- passed, no must-fix
findings
References:
None - regression coverage for async boundary batching
---
.../test/scala/org/apache/pekko/stream/FusingSpec.scala | 16 ++++++++++++++++
1 file changed, 16 insertions(+)
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 03648f78c5..1610f1c3ff 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
@@ -133,6 +133,22 @@ class FusingSpec extends StreamSpec {
downstream.expectError(ex)
}
+ "deliver transformed elements already produced before an asynchronous
boundary failure" in {
+ val ex = TE("mapping failed after partial output")
+
+ Source(Vector("amber", "birch", "cobalt", "dune", "fault-line"))
+ .map {
+ case "fault-line" => throw ex
+ case element => element.reverse
+ }
+ .async
+ .addAttributes(asyncBoundaryInputBuffer)
+ .runWith(TestSink[String]())
+ .request(8)
+ .expectNext("rebma", "hcrib", "tlaboc", "enud")
+ .expectError(ex)
+ }
+
"not emit elements after an asynchronous boundary failure" in {
val ex = TE("boom")
val probe = Source(1 to 64)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
