You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2026/01/19 02:57:28 UTC
[PR] fix: Fix recoverWith on Failed stage. [pekko]
He-Pin opened a new pull request, #2631:
URL: https://github.com/apache/pekko/pull/2631
Motivation:
Fix https://github.com/apache/pekko/issues/2620
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix recoverWith on Failed stage. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #2631:
URL: https://github.com/apache/pekko/pull/2631#discussion_r2703093901
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceSpec.scala:
##########
@@ -715,33 +715,30 @@ class SourceSpec extends StreamSpec with DefaultTimeout {
assert(counter.get() == 3)
}
- "not retry FailedSources" in {
- // https://github.com/apache/pekko/issues/2620
+ "should retry on a failed source" in {
val counter = new java.util.concurrent.atomic.AtomicInteger()
val source =
- withRetriesTest(failedSource("origin")) { _ =>
- counter.incrementAndGet()
- failedSource("does not work")
- } { _ =>
- counter.get() < 3
- }
+ withRetriesTest(failedSource("origin")) { () =>
+ if (counter.incrementAndGet() < 3) {
+ failedSource("does not work")
+ } else Source.single(ByteString.fromString("ok"))
+ } { _ => true }
+ .runWith(Sink.head)
+ val result = Await.result(source, Duration.Inf)
+ assert(result.utf8String == "ok")
- assertThrows[ArithmeticException] {
- Await.result(source.runWith(Sink.ignore), Duration.Inf)
- }
-
- assert(counter.get() == 1)
+ assert(counter.get() == 3)
Review Comment:
Updated this with fix
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix recoverWith on Failed stage. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #2631:
URL: https://github.com/apache/pekko/pull/2631#discussion_r2703094654
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala:
##########
@@ -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)
Review Comment:
the fix is here too
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix recoverWith on Failed stage. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #2631:
URL: https://github.com/apache/pekko/pull/2631
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix recoverWith on Failed stage. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #2631:
URL: https://github.com/apache/pekko/pull/2631#discussion_r2703094362
##########
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala:
##########
@@ -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)
Review Comment:
The fix is here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix recoverWith on Failed stage. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #2631:
URL: https://github.com/apache/pekko/pull/2631#issuecomment-3766234292
@pjfanning Seems I need backport this to both 1.5.x and 1.4.x?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org