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 2024/01/19 06:05:12 UTC
[I] Feature request: Resume with Error logging for Stream [incubator-pekko]
He-Pin opened a new issue, #980:
URL: https://github.com/apache/incubator-pekko/issues/980
Motivation:
I saw in https://github.com/akka/akka/issues/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.
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
--
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.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: [I] Feature request: Resume with Error logging for Stream [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900088020
@mdedetrich But your proposal can not swith to an `ignore` sink.
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [incubator-pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900105790
> 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`](https://pekko.apache.org/docs/pekko/current/stream/operators/Partition.html) 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 https://github.com/apache/incubator-pekko/pull/981
--
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: [I] Feature request: Resume with Error logging for Stream [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1899850621
Yes, I was expecting some factory method, eg `Decider.resume(level:LoggingLevel)`, let's wait for some feedbacks form @tg44 and @nvollmar
--
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: [I] Feature request: Resume with Error logging for Stream [incubator-pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900079808
@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
```scala
val restartingDecider: Supervision.Decider = { e =>
logger.error("Unhandled exception in stream, restarting", e)
Supervision.Restart
}
source
.splitAfter(_ == somePredicate)
.withAttributes(ActorAttributes.supervisionStrategy(restartingDecider))
```
--
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: [I] Feature request: Resume with Error logging for Stream [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900072239
@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`.
```scala
Data --> 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.
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900826547
Yes, I love your contritions @tg44 and I used it too.
And I think what's you did is some kind of *Isolate strong and weak dependencies to prevent strong dependencies from being affected by weak dependencies*
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [incubator-pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900093480
@He-Pin https://github.com/apache/incubator-pekko/pull/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`](https://pekko.apache.org/docs/pekko/current/stream/operators/Partition.html)
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [incubator-pekko]
Posted by "tg44 (via GitHub)" <gi...@apache.org>.
tg44 commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900213653
@mdedetrich Nah, when I started to work with akka streams we tend to call every "part of the stream after a fanout" is a substream, and `groupBy` and the "real" `substream`/`subflow` as a term was introduced/standardized later on. (I was a really really early adopter, I think I have merged PRs almost every akka repos (akka, akka-http, I wrote/rewrote components to alpakka).) So sorry for the misunderstanding.
For me the fileIO is kinda a log to a file, so I choose wireTap with an intention, I have buffers and logrotator and everything in between the wireTap and the actual fileIO, I can allow to not write everything to the fileLog, but I need to write everything to the database as fast as possible.
--
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: [I] Feature request: Resume with Error logging for Stream [incubator-pekko]
Posted by "tg44 (via GitHub)" <gi...@apache.org>.
tg44 commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900032628
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.
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [incubator-pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1901200379
Okay so in the end this seems to be a request for an easier DSL for `Sink` and while https://github.com/apache/incubator-pekko/pull/981 is valid its also not related.
Now that I think of it, this honestly seems to be asking for a `Sink` equivalent of [`RetryFlow.withBackoff`](https://doc.akka.io/docs/akka/current/stream/operators/RetryFlow/withBackoff.html), @He-Pin wdyt
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on issue #980:
URL: https://github.com/apache/pekko/issues/980#issuecomment-3064813093
This require large refactory to RestartWithBackoffLogic
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900104147
An out of box seems better, I think some kind of ZIO ZSink's `orElse` but only switch after some kind of maxRetriesWithBackOff.
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1901227568
@mdedetrich That's true, but we can extends to `Retry/Restart*.withBackOff/fallback`, that's what I have in mind.
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [incubator-pekko]
Posted by "tg44 (via GitHub)" <gi...@apache.org>.
tg44 commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900165463
My usecase looks like this;
```
Data Source X
|
\|/
Datas
|
\|/
Process A
|
\|/
wireTap()----> Process FileIO Sink (main, can retry with backoff restart)
| | (if it is restarted with supervision it will fail at every retry effectively killing the database save)
\|/ | (only switch after the main sink die and restart up to the max limit)
database save |
|----> fallback Ignore Sink(logging) maybe.
```
I think we don't have a Sink which can handle the right side of the graph. It could be something like;
```scala
def sinkWithFallback[T, Mat](factory: () => (exception: Option[Throwable]) => Sink[T, Mat]): Sink[T, Mat]
```
And it should be easily implemented from `lazySink`, but `lazySink` is internal (or at least the akka version I'm using rn has it in internal), so it is messy...
--
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: [I] Feature request: Resume with Error logging for Stream [incubator-pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1899826294
You can do this already due to this change https://github.com/apache/incubator-pekko/pull/252 which has already landed.
You can just make your own supervision decider, i.e.
```scala
val decider: Supervision.Decider = { e =>
logger.error("Unhandled exception in stream", e)
Supervision.Stop
}
```
or am I missing something?
--
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: [I] Feature request: Resume with Error logging for Stream [incubator-pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900053232
@tg44 So if by Sub Stream you mean `SubFlow` the issue you describe is partially solved in Pekko 1.1.x with this [PR](https://github.com/apache/incubator-pekko/pull/252) (which isn't release yet but there are snapshots if you want to try). You can read the migration notes [here](https://github.com/apache/incubator-pekko/blob/32c5d2c306e8d7dba78befaa0a09b5b1f62d66e7/docs/src/main/paradox/project/migration-guide-1.0.x-1.1.x.md#sourcesplitwhenflowsplitwhensourcesplitafterflowsplitafter-behavioural-change)
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
@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
--
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: [I] Feature request: RestartWithBackOffAndFallback for Flow & Sink [incubator-pekko]
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on issue #980:
URL: https://github.com/apache/incubator-pekko/issues/980#issuecomment-1900179311
@tg44 So just to confirm, buy sub stream you didn't mean `split`/`when` (which creates a `SubFlow`) but instead `wireTap()`.
Also is there a reason why you aren't using `alsoTo` instead of `wireTap`? Usually for your scenario you would use `alsoTo` since `wireTap` will not backpressure the flow it is tapped into (hence `wireTap` is typically only used for logging/metrics and the such)
--
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