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 efc254b8b0 Fix MapAsyncPartitioned demand-resume deadlock #2886 (#3365)
efc254b8b0 is described below
commit efc254b8b0d9fa3b64f614e737420906537bdb10
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jul 19 04:21:30 2026 +0800
Fix MapAsyncPartitioned demand-resume deadlock #2886 (#3365)
Motivation:
Completed mapAsyncPartitioned results can remain buffered until downstream
demand resumes. The partitionsInProgress empty shortcut bypassed emission and
could leave the stage unable to make progress.
Modification:
Always evaluate buffered completed elements in the ordered and unordered
push paths. Add focused backpressure tests for both variants.
Result:
Completed elements are emitted when downstream demand resumes without
changing ordering or partition-parallelism semantics.
Tests:
- stream-typed-tests MapAsyncPartitionedSpec: 5 repeated runs, 110/110
passed
- stream-tests FlowMapAsyncPartitionedSpec: 24/24 passed
- scalafmt diff mode and list check: passed
- headerCreateAll, checkCodeStyle, and +headerCheckAll: passed
- validatePullRequest: relevant stream batches passed; overall failed on
unrelated macOS ARM64/JDK 25 environment tests documented in the PR
- git diff --check: passed
- Qoder stdout review: No must-fix findings
References:
Fixes #2886
---
.../scaladsl/FlowMapAsyncPartitionedSpec.scala | 41 ++++++++
.../apache/pekko/stream/MapAsyncPartitioned.scala | 108 ++++++++++-----------
2 files changed, 91 insertions(+), 58 deletions(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncPartitionedSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncPartitionedSpec.scala
index 1920af03c2..6d158a80ce 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncPartitionedSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowMapAsyncPartitionedSpec.scala
@@ -54,6 +54,39 @@ class FlowMapAsyncPartitionedSpec extends StreamSpec with
WithLogCapturing {
override implicit def patienceConfig: PatienceConfig =
PatienceConfig(timeout = 30.seconds.dilated, interval = Span(15, Millis))
+ private def
verifyCompletedElementsAreEmittedAfterDemandResumes(orderedOutput: Boolean):
Unit = {
+ val processingProbe = TestProbe()
+ val promises = Array.fill(4)(Promise[Int]())
+ val operation: (Int, Int) => Future[Int] = { (elem, _) =>
+ processingProbe.ref ! elem
+ promises(elem).future
+ }
+ val flow =
+ if (orderedOutput) Flow[Int].mapAsyncPartitioned(3)(identity)(operation)
+ else Flow[Int].mapAsyncPartitionedUnordered(3)(identity)(operation)
+ val downstream = Source(0 until 4).via(flow).runWith(TestSink[Int]())
+
+ downstream.request(1)
+ processingProbe.expectMsg(0)
+ processingProbe.expectMsg(1)
+ processingProbe.expectMsg(2)
+
+ promises(0).success(0)
+ downstream.expectNext(0)
+ processingProbe.expectMsg(3)
+
+ promises(1).success(1)
+ promises(2).success(2)
+ promises(3).success(3)
+ downstream.expectNoMessage(200.millis)
+
+ downstream.request(3)
+ val remaining = downstream.expectNextN(3)
+ if (orderedOutput) remaining should ===(Seq(1, 2, 3))
+ else remaining should contain theSameElementsAs Seq(1, 2, 3)
+ downstream.expectComplete()
+ }
+
"A Flow with mapAsyncPartitioned" must {
"produce future elements" in {
implicit val ec: ExecutionContext = system.dispatcher
@@ -110,6 +143,14 @@ class FlowMapAsyncPartitionedSpec extends StreamSpec with
WithLogCapturing {
c.expectComplete()
}
+ "emit completed ordered elements after downstream demand resumes" in {
+ verifyCompletedElementsAreEmittedAfterDemandResumes(orderedOutput = true)
+ }
+
+ "emit completed unordered elements after downstream demand resumes" in {
+ verifyCompletedElementsAreEmittedAfterDemandResumes(orderedOutput =
false)
+ }
+
"not run more futures than overall parallelism" in {
implicit val ec: ExecutionContext = system.dispatcher
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/MapAsyncPartitioned.scala
b/stream/src/main/scala/org/apache/pekko/stream/MapAsyncPartitioned.scala
index b7ca9219d2..790551f79d 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/MapAsyncPartitioned.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/MapAsyncPartitioned.scala
@@ -205,24 +205,58 @@ private[stream] final class MapAsyncPartitioned[In, Out,
Partition](
if (orderedOutput) pushNextIfPossibleOrdered _
else pushNextIfPossibleUnordered _
- private def pushNextIfPossibleOrdered(): Unit =
- if (partitionsInProgress.isEmpty) {
- drainQueue()
- pullIfNeeded()
- } else {
- while (buffer.nonEmpty && !(buffer.front._2.element.out eq
NotYetThere) && isAvailable(out)) {
- val (partition, wrappedInput) = buffer.dequeue()
- import wrappedInput.{ element => holder }
+ private def pushNextIfPossibleOrdered(): Unit = {
+ while (buffer.nonEmpty && !(buffer.front._2.element.out eq
NotYetThere) && isAvailable(out)) {
+ val (partition, wrappedInput) = buffer.dequeue()
+ import wrappedInput.{ element => holder }
+ partitionsInProgress -= partition
+
+ holder.out match {
+ case Success(elem) =>
+ if (elem != null) {
+ push(out, elem)
+ pullIfNeeded()
+ } else {
+ // elem is null
+ pullIfNeeded()
+ }
+
+ case Failure(NonFatal(ex)) =>
+ holder.supervisionDirectiveFor(decider, ex) match {
+ // this could happen if we are looping in pushNextIfPossible
and end up on a failed future before the
+ // onComplete callback has run
+ case Supervision.Stop =>
+ failStage(ex)
+ case _ =>
+ // try next element
+ logSupervisionFailure(ex)
+ }
+ case Failure(ex) =>
+ // fatal exception in buffer, not sure that it can actually
happen, but for good measure
+ throw ex
+ }
+ }
+ drainQueue()
+ // The while-loop's Failure (resume) branch dequeues elements without
calling pullIfNeeded(),
+ // so when the final buffered elements are resumed failures the buffer
empties here without any
+ // completion check. pullIfNeeded() is the only path that runs
completeStage() once upstream has
+ // finished, so it must run after draining or the stage hangs forever
(mirrors the unordered branch).
+ pullIfNeeded()
+ }
+
+ private def pushNextIfPossibleUnordered(): Unit = {
+ buffer = buffer.filter { case (partition, wrappedInput) =>
+ import wrappedInput.{ element => holder }
+
+ if ((holder.out eq NotYetThere) || !isAvailable(out)) {
+ true
+ } else {
partitionsInProgress -= partition
holder.out match {
case Success(elem) =>
if (elem != null) {
push(out, elem)
- pullIfNeeded()
- } else {
- // elem is null
- pullIfNeeded()
}
case Failure(NonFatal(ex)) =>
@@ -239,54 +273,12 @@ private[stream] final class MapAsyncPartitioned[In, Out,
Partition](
// fatal exception in buffer, not sure that it can actually
happen, but for good measure
throw ex
}
+ false
}
- drainQueue()
- // The while-loop's Failure (resume) branch dequeues elements
without calling pullIfNeeded(),
- // so when the final buffered elements are resumed failures the
buffer empties here without any
- // completion check. pullIfNeeded() is the only path that runs
completeStage() once upstream has
- // finished, so it must run after draining or the stage hangs
forever (mirrors the unordered branch).
- pullIfNeeded()
- }
-
- private def pushNextIfPossibleUnordered(): Unit =
- if (partitionsInProgress.isEmpty) {
- drainQueue()
- pullIfNeeded()
- } else {
- buffer = buffer.filter { case (partition, wrappedInput) =>
- import wrappedInput.{ element => holder }
-
- if ((holder.out eq NotYetThere) || !isAvailable(out)) {
- true
- } else {
- partitionsInProgress -= partition
-
- holder.out match {
- case Success(elem) =>
- if (elem != null) {
- push(out, elem)
- }
-
- case Failure(NonFatal(ex)) =>
- holder.supervisionDirectiveFor(decider, ex) match {
- // this could happen if we are looping in
pushNextIfPossible and end up on a failed future before the
- // onComplete callback has run
- case Supervision.Stop =>
- failStage(ex)
- case _ =>
- // try next element
- logSupervisionFailure(ex)
- }
- case Failure(ex) =>
- // fatal exception in buffer, not sure that it can actually
happen, but for good measure
- throw ex
- }
- false
- }
- }
- pullIfNeeded()
- drainQueue()
}
+ pullIfNeeded()
+ drainQueue()
+ }
private def drainQueue(): Unit = {
if (buffer.nonEmpty) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
