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 c532ea043b fix: delay unfoldResourceAsync close until pending read
completes (#3157)
c532ea043b is described below
commit c532ea043b9d16ef365b40dc63234b27fc71dc4e
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Wed Jun 24 13:01:48 2026 +0800
fix: delay unfoldResourceAsync close until pending read completes (#3157)
Motivation:
Source.unfoldResourceAsync could call close(resource) from postStop while a
readData(resource) Future was still running, allowing close and read to operate
on the same resource concurrently.
Modification:
Track the current read Future with a null-sentinel field and, on postStop,
defer close(resource) until that Future completes. Clear the field on
synchronous and asynchronous read completion to keep the hot path simple and
avoid retaining completed reads. Add a regression test that cancels while a
read is pending and asserts close is not called until the read completes.
Result:
Cancellation no longer closes the resource concurrently with an in-flight
readData call, while existing unfoldResourceAsync completion, restart, resume,
and close behavior remains covered.
Tests:
- scalafmt --mode diff-ref=origin/main: passed (JDK 25 Unsafe deprecation
warning only)
- scalafmt --list --mode diff-ref=origin/main: passed (JDK 25 Unsafe
deprecation warning only)
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.UnfoldResourceAsyncSourceSpec -- -z \"wait for
pending read before closing resource when stream is cancelled\"": passed
- same focused test against old postStop behavior: failed as expected with
close observing readCompleted=false
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.UnfoldResourceAsyncSourceSpec": passed, 21
tests
- sbt checkCodeStyle: passed
- sbt +headerCheckAll: passed
- git diff --check: passed
- sbt validatePullRequest: Not run - focused stream test, style, and header
validation run locally
References:
Fixes #3134
---
.../scaladsl/UnfoldResourceAsyncSourceSpec.scala | 34 ++++++++++++++++++++++
.../stream/impl/UnfoldResourceSourceAsync.scala | 21 +++++++++----
2 files changed, 50 insertions(+), 5 deletions(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/UnfoldResourceAsyncSourceSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/UnfoldResourceAsyncSourceSpec.scala
index 07a42be8d8..829762e0ac 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/UnfoldResourceAsyncSourceSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/UnfoldResourceAsyncSourceSpec.scala
@@ -129,6 +129,40 @@ class UnfoldResourceAsyncSourceSpec extends
StreamSpec(UnboundedMailboxConfig) {
resource.closed.futureValue
}
+ "wait for pending read before closing resource when stream is cancelled"
in {
+ val readStarted = Promise[Done]()
+ val readCompleted = Promise[Option[Int]]()
+ val closeProbe = TestProbe()
+ val out = TestSubscriber.manualProbe[Int]()
+
+ val terminated = Source
+ .unfoldResourceAsync[Int, Unit](
+ () => Future.successful(()),
+ _ => {
+ readStarted.success(Done)
+ readCompleted.future
+ },
+ _ => {
+ closeProbe.ref ! readCompleted.isCompleted
+ Future.successful(Done)
+ })
+ .watchTermination(Keep.right)
+ .toMat(Sink.fromSubscriber(out))(Keep.left)
+ .run()
+
+ val subscription = out.expectSubscription()
+ subscription.request(1)
+ readStarted.future.futureValue
+
+ subscription.cancel()
+ terminated.futureValue
+ closeProbe.expectNoMessage(200.millis)
+
+ readCompleted.success(Some(1))
+ closeProbe.expectMsg(true)
+ closeProbe.expectNoMessage(200.millis)
+ }
+
"fail when create throws exception" in {
val probe = TestSubscriber.probe[Unit]()
Source
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/UnfoldResourceSourceAsync.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/UnfoldResourceSourceAsync.scala
index b0a74bfcb7..fff8b10a58 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/UnfoldResourceSourceAsync.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/UnfoldResourceSourceAsync.scala
@@ -46,6 +46,7 @@ import pekko.util.OptionVal
private lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider
private implicit def ec: ExecutionContext = materializer.executionContext
private var maybeResource: OptionVal[R] = OptionVal.none
+ private var inFlightRead: Future[Option[T]] = _
private val createdCallback = getAsyncCallback[Try[R]] {
case Success(resource) =>
@@ -70,7 +71,10 @@ import pekko.util.OptionVal
}
}
- private val readCallback =
getAsyncCallback[Try[Option[T]]](handle).invoke _
+ private val readCallback = getAsyncCallback[Try[Option[T]]] { result =>
+ inFlightRead = null
+ handle(result)
+ }.invoke _
private def handle(result: Try[Option[T]]): Unit = result match {
case Success(data) =>
@@ -100,9 +104,12 @@ import pekko.util.OptionVal
case OptionVal.Some(resource) =>
try {
val future = readData(resource)
+ inFlightRead = future
future.value match {
- case Some(value) => handle(value)
- case None => future.onComplete(readCallback)(parasitic)
+ case Some(value) =>
+ inFlightRead = null
+ handle(value)
+ case None => future.onComplete(readCallback)(parasitic)
}
} catch errorHandler
@@ -113,8 +120,12 @@ import pekko.util.OptionVal
}
override def postStop(): Unit = maybeResource match {
- case OptionVal.Some(resource) => close(resource)
- case _ => // do nothing
+ case OptionVal.Some(resource) =>
+ val pendingRead = inFlightRead
+ inFlightRead = null
+ if (pendingRead ne null) pendingRead.onComplete(_ =>
close(resource))(parasitic)
+ else close(resource)
+ case _ => // do nothing
}
private def restartResource(): Unit = maybeResource match {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
