This is an automated email from the ASF dual-hosted git repository. hepin pushed a commit to branch recoverWithFix in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 022592131a52e433c96a21c2244ce5e49316906d Author: He-Pin <[email protected]> AuthorDate: Mon Jan 19 10:56:49 2026 +0800 fix: Fix recoverWith on Failed stage. --- .../stream/scaladsl/FlowRecoverWithSpec.scala | 36 +++++++++++++++++++--- .../org/apache/pekko/stream/impl/fusing/Ops.scala | 5 +-- .../org/apache/pekko/stream/javadsl/Flow.scala | 4 +++ .../org/apache/pekko/stream/javadsl/Source.scala | 4 +++ .../org/apache/pekko/stream/javadsl/SubFlow.scala | 2 ++ .../apache/pekko/stream/javadsl/SubSource.scala | 2 ++ .../org/apache/pekko/stream/scaladsl/Flow.scala | 2 ++ 7 files changed, 49 insertions(+), 6 deletions(-) diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowRecoverWithSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowRecoverWithSpec.scala index 2ba9ed4a51..47f129a329 100644 --- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowRecoverWithSpec.scala +++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowRecoverWithSpec.scala @@ -73,11 +73,39 @@ class FlowRecoverWithSpec extends StreamSpec { } "recover with a failed future source" in { - Source.failed(ex) - .recoverWith { case _: Throwable => Source.future(Future.failed(ex)) } + val counter = new java.util.concurrent.atomic.AtomicInteger(0) + Source.failed[Int](ex) + .recoverWith { + case _: Throwable => + if (counter.incrementAndGet() < 100) { + Source.future(Future.failed(ex)) + } else { + Source.single(101) + } + } .runWith(TestSink[Int]()) - .request(1) - .expectError(ex) + .request(100) + .expectNext(101) + .expectComplete() + counter.get() shouldBe 100 + } + + "recover with a failed source" in { + val counter = new java.util.concurrent.atomic.AtomicInteger(0) + Source.failed[Int](ex) + .recoverWith { + case _: Throwable => + if (counter.incrementAndGet() < 100) { + Source.failed(ex) + } else { + Source.single(101) + } + } + .runWith(TestSink[Int]()) + .request(100) + .expectNext(101) + .expectComplete() + counter.get() shouldBe 100 } "recover with a java stream source" in { diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala index 352bd64fa7..d02744c8d3 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala @@ -2169,6 +2169,7 @@ private[pekko] object TakeWithin { override def onPull(): Unit = pull(in) @nowarn("msg=Any") + @tailrec def onFailure(ex: Throwable): Unit = { import Collect.NotApplied if (maximumRetries < 0 || attempt < maximumRetries) { @@ -2180,10 +2181,10 @@ private[pekko] object TakeWithin { TraversalBuilder.getValuePresentedSource(source) match { case OptionVal.Some(graph) => graph match { case singleSource: SingleSource[T @unchecked] => emit(out, singleSource.elem, () => completeStage()) - case failed: FailedSource[T @unchecked] => failStage(failed.failure) + case failed: FailedSource[T @unchecked] => onFailure(failed.failure) case futureSource: FutureSource[T @unchecked] => futureSource.future.value match { case Some(Success(elem)) => emit(out, elem, () => completeStage()) - case Some(Failure(ex)) => failStage(ex) + case Some(Failure(ex)) => onFailure(ex) case None => switchTo(source) attempt += 1 diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala index 88f40cee84..1622dad4c6 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala @@ -2078,6 +2078,8 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr * * Throwing an exception inside `recoverWith` _will_ be logged on ERROR level automatically. * + * It will keep trying to recover indefinitely, if you want to limit the number of attempts, use `recoverWithRetries`. + * * '''Emits when''' element is available from the upstream or upstream is failed and element is available * from alternative Source * @@ -2100,6 +2102,8 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr * * Throwing an exception inside `recoverWith` _will_ be logged on ERROR level automatically. * + * It will keep trying to recover indefinitely, if you want to limit the number of attempts, use `recoverWithRetries`. + * * '''Emits when''' element is available from the upstream or upstream is failed and element is available * from alternative Source * diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala index 40c41647f0..cd045591a5 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala @@ -2317,6 +2317,8 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[ * * Throwing an exception inside `recoverWith` _will_ be logged on ERROR level automatically. * + * It will keep trying to recover indefinitely, if you want to limit the number of attempts, use `recoverWithRetries`. + * * '''Emits when''' element is available from the upstream or upstream is failed and element is available * from alternative Source * @@ -2339,6 +2341,8 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[ * * Throwing an exception inside `recoverWith` _will_ be logged on ERROR level automatically. * + * It will keep trying to recover indefinitely, if you want to limit the number of attempts, use `recoverWithRetries`. + * * '''Emits when''' element is available from the upstream or upstream is failed and element is available * from alternative Source * diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala index fa3ee316b1..14f525bb70 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala @@ -1427,6 +1427,8 @@ final class SubFlow[In, Out, Mat]( * * Throwing an exception inside ``recoverWith`` _will_ be logged on ERROR level automatically. * + * It will keep trying to recover indefinitely, if you want to limit the number of attempts, use `recoverWithRetries`. + * * '''Emits when''' element is available from the upstream or upstream is failed and element is available * from alternative Source * diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala index 6324c447f9..7edb90afd9 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala @@ -1399,6 +1399,8 @@ final class SubSource[Out, Mat]( * Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. * This operator can recover the failure signal, but not the skipped elements, which will be dropped. * + * It will keep trying to recover indefinitely, if you want to limit the number of attempts, use `recoverWithRetries`. + * * '''Emits when''' element is available from the upstream or upstream is failed and element is available * from alternative Source * diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala index 00c7cec924..49d0c8382e 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala @@ -906,6 +906,8 @@ trait FlowOps[+Out, +Mat] { * * Throwing an exception inside `recoverWith` _will_ be logged on ERROR level automatically. * + * It will keep trying to recover indefinitely, if you want to limit the number of attempts, use `recoverWithRetries`. + * * '''Emits when''' element is available from the upstream or upstream is failed and element is available * from alternative Source * --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
