RecoverWith operator break of semantics in 1.2.0 release

[kstokoz]

I was moving my project from pekko 1.0.1 to 1.2.0. One of my test which using source.recoverWithRetries failed.

I believe in change #1775 and #1663 merged to pekko 1.2.0 release in case of Source.failed semantics of the API is actualy changed, i.e. max attempts of -1 is no longer working properly.

Please consider following test:

import java.util.concurrent.atomic.AtomicInteger
import org.apache.pekko.stream.scaladsl.Sink
import org.apache.pekko.stream.scaladsl.Source
import org.apache.pekko.util.ByteString
import org.apache.pekko.NotUsed

import scala.concurrent.Await
import scala.concurrent.duration.Duration

class ResilientSourceTest extends UnitTest {

  def withRetriesTest(originSource: Source[ByteString, Any])(fallbackTo: Long => Source[ByteString, NotUsed])(shouldRetry: Throwable => Boolean = { _ => true }): Source[ByteString, NotUsed] =
    originSource.recoverWithRetries(
      -1,
      {
        case e: Throwable if shouldRetry(e) =>
          fallbackTo(0)
      }
    ).mapMaterializedValue(_ => NotUsed)

  test("withRetries should stop to retry if shouldRetry decides otherwise") {
    val counter = new AtomicInteger()

    val source =
      withRetriesTest(failedSource("origin")) { _ =>
        counter.incrementAndGet()
//        failedSource1("works")
        failedSource("does not work")
      } { _ =>
        counter.get() < 3
      }

    assertThrows[ArithmeticException] {
      Await.result(source.runWith(Sink.ignore), Duration.Inf)
    }

    assert(counter.get() == 3)
  }

  private def failedSource(message: String): Source[ByteString, NotUsed] =
    Source.failed(new ArithmeticException(message))

  private def failedSource1(message: String): Source[ByteString, NotUsed] =
    Source.single(5).map(_ / 0).map(s => ByteString.fromString(s.toString))
}

When i have Source.failed (case failedSource) test is not passing, while i would expect it should, similar if source would produce error during execution (case failedSource1)

Regards,
Kyrylo

[pjfanning]

recoverWithRetries is documented to support negative retries and means infinite retries so this looks like a regression (and a bug).

https://pekko.apache.org/docs/pekko/current/stream/operators/Source-or-Flow/recoverWithRetries.html

wdyt @He-Pin ?

[pjfanning]

I have reproduced the issue. I'm not sure whether we should revert 1775 or try to fix the issue that it introduces with retrying where Source.failed occurs (failedSource in the example).

@kstokoz it is a very artificial example that is failing. In the real world, why would you want to retry a Source.failed?

[kstokoz]

@pjfanning it is indeed a test scenario where i want to test retry logic, having Source.failed is defined method to create failing source. Even though it is a test i more worried about sematics change of the API, when failed source created via different options.
I don't have test for other cases in the change like futureSource or javaSource, but i don't think assumption that we can shortcut some logic is safe here.

I would vote for reverting the original change.

[pjfanning]

Pekko 1.2.0 is about 4 months old and nobody else has reported issues like this. I would argue that we need a better test case. I'm not fully against a revert but I wouldn't expedite a release without a more worrying scenario.

[pjfanning]

My gut feeling is that

pekko/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala

Line 2183 in 4479947

case failed: FailedSource[T @unchecked] => failStage(failed.failure)

is safe - that we don't retry if the source is an instance of FailedSource and for me, this seems sensible.

It would be one thing if the source was anything other a FailedSource, a source that had a chance to succeed - but an instanceof FailedSource, it seems to me, that it is an improvement not to retry.

[kstokoz]

If you think that current behavior is an optimisation and it would keep failing right away on failed source, i think you should document this behavior. I should admit it would very confusing for users that when you set -1 for infinite retries and it doesn`t do it in some cases. Any user who want to provoke error in test or in general will use failed source.

[pjfanning]

Whatever gets agreed, we will make sure the docs reflect it.

I'm waiting for other Pekko contributors to voice an opinion before choosing whether to revert the 1.2.0 change or to keep it and document this behaviour change.

[He-Pin]

I was in the hospital, and I will take a look this weekend.
@pjfanning @kstokoz Sorry for the delay, I was thinking that Explicit failures should not be retried.

[He-Pin]

@kstokoz @pjfanning I think we should update the document about this and with a test case in #2624

[pjfanning]

@He-Pin @raboof @mdedetrich the more I think about this, the more I think that we should fix this. The recover functions take as one param a function to create a recovery flow. That function could return a hard coded FailedSource in an artificial test but it could have logic that depending on the input value or external factor if the function allows that then it could return a FailedSource or a viable Source. For example first retry could lead to a FailedSource but 2nd could lead to a different type of Source that is worth trying.

[He-Pin]

A problem with that is: if the user is not using recoverWithRetries, it will retry in an infinite loop.
Even the fix is quite simple.

Maybe we can add a warning per 100 retrie times?