Feature request: RestartWithBackOffAndFallback for Flow & Sink

[He-Pin]

Motivation:
I saw in akka/akka-core#30267 @nvollmar added this feature for actor, but I think it would be nice to have this in pekko-stream, where user can logging the error with a restart/resume, and currently ,we just simply ignore it.

I found the related case from @tg44 in https://discuss.lightbend.com/t/recover-from-sink-exceptions/10554

I think this should be very useful for real production usecase.

Result:
More handy control over the exceptions

[mdedetrich]

You can do this already due to this change #252 which has already landed.

You can just make your own supervision decider, i.e.

val decider: Supervision.Decider = { e =>
  logger.error("Unhandled exception in stream", e)
  Supervision.Stop
}

or am I missing something?

[He-Pin]

Yes, I was expecting some factory method, eg Decider.resume(level:LoggingLevel), let's wait for some feedbacks form @tg44 and @nvollmar

[tg44]

So, my initial problem was that I have a substream which has a fileIO in it, the fileIO was only a safety measure, so if we loose the db connection we don't loose (too much) data. Then one day the fileIO failed (we couldn't recreate it bcs it opened a file which failed due to hosting provider issues), and it brought down the whole stream, and we lost hours of data when we finally noticed that the stream is down...

My idea was to lazily create a sink, if it fails I can recreate it, if it fails multiple times I can simply change it to a Sink.ignore if I want to.

[mdedetrich]

@tg44 So if by Sub Stream you mean SubFlow the issue you describe is partially solved in Pekko 1.1.x with this PR (which isn't released yet but there are snapshots if you want to try). You can read the migration notes here

What I mean by partially is that as is noted in the migration notes we don't support Supervision.restart. Initially when implementing that PR I had difficulties supporting Supervision.restart but since there seems to be some demand for it I can look into it if you confirm that Supervision.restart support for SubFlow's is indeed what you are actually asking for.

@He-Pin Aside from what I just said I don't think we need to add anything else for Pekko, it appears that @tg44 just wants support for automatic restarting of SubFlow's

[He-Pin]

@mdedetrich I think what's @tg44 needs is :

1. a RestartSink, with is currently missing, maybe that can be done with the RestartFlow.to(IgnoreSink)
2. a fallback, but for the sink, the main sink was a fileIO sink ,but after some max retry, switch to a Sink.ignore.

Data Source X
   |
  \|/
Datas --> Process A ----> Process FileIO Sink (main, can retry with backoff restart)
                                      |
                                      |  (only switch after the main sink die and restart up to the max limit)
                                      |
                                      |----> fallback Ignore Sink(logging) maybe.
           

And I think this is what he really want.

[mdedetrich]

@He-Pin But he is talking about SubFlow, so from what he seems to be saying is that he just wants the SubFlow to have the ability to restart (and this means by implication if the SubFlow is just a sink then the Sink will restart).

Then you can just do

val restartingDecider: Supervision.Decider = { e =>
  logger.error("Unhandled exception in stream, restarting", e)
  Supervision.Restart
}

source
  .splitAfter(_ == somePredicate)
  .withAttributes(ActorAttributes.supervisionStrategy(restartingDecider))

[He-Pin]

@mdedetrich But your proposal can not swith to an ignore sink.

I see you just updated a pr, what about extends RestartFlow to support a fallback too.

[mdedetrich]

@He-Pin #981 is what I was thinking of (still needs to be tested, there may be some mistake)

But your proposal can not swith to an ignore sink.

Its not necessary though, you can do this yourself by making a custom sink with Partition?

[He-Pin]

An out of box seems better, I think some kind of ZIO ZSink's orElse but only switch after some kind of maxRetriesWithBackOff.

[mdedetrich]

An out of box seems better, I think some kind of ZIO ZSink's orElse but only switch after some kind of maxRetriesWithBackOff.

Yeah using Partition right now is a bit clunky because of how manual it is but thats another topic

@tg44 Let me know if I am on the right track and if so I will work on #981