SubFlow's don't correctly propagate Supervision.resumeStrategy

[mdedetrich]

In #252 I made it so that we can set SupervisionStrategy rather than using SubStreamCancelStrategy. It turns out there is an existing test which was written as pending (so its currently disabled) to test this exact SupervisionStrategy functionality (specifically SupervisionStrategy.resumingDecider when an exception is thrown) when its adding in the future (which is now) i.e.

https://github.com/apache/incubator-pekko/blob/c44c0b7cbdab11d85176cfe062288fdcba16c56a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitAfterSpec.scala#L213-L262

When I try to enable the test by removing pending so the test runs it actually does not pass which means that the SupervisionStrategy.resumingDecider is not properly catching and suppressing the exception as its meant to do so.

[mdedetrich]

So I set the 1.1.0-M1 milestone for this bug, will do my best to solve it by the time the -M1 release is actually cut however if its not done by then can do it in a following -M2 or full release.

It does however need to be solved by full 1.1.0 release

[gwak]

Hi @mdedetrich,

@pjfanning and I just tested the 1.1 milestone release with playframework. We followed the migration guide and explicitly defined the supervision strategy when using the splitWhen function like so:

.splitWhen(_.isLeft)
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
.prefixAndTail(1)

But when using the deprecated SubstreamCancelStrategy it works fine:

.splitWhen(SubstreamCancelStrategy.drain)(_.isLeft)

Is there something we're missing ?

[mdedetrich]

@gwak So I still need to work on this, in the meantime I would recommend using the deprecated method as you described.

I just skimmed through the issue to recover some context, iirc the core issue is that in general ActorAttributes are not propagating where as in Pekko 1.0 and/or current pekko 1.1 deprecation functionality propagates the strategy in a different/more explicit way

[mdedetrich]

Re-opening this as it needs to be solved

[mdedetrich]

@gwak So I am diving into this now and I think that I fixed this issue in #1207 so things should be working as expected.

Reading your code example at surface level, maybe the problem is that you are not doing the equivalent of .lift which an be seen here

pekko/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowSplitAfterSpec.scala

Line 219 in 689e30b

.lift

, you can see the definition of lift here

[gwak]

@mdedetrich You were right, @pjfanning tried your suggestion in playframework/playframework#12794 and tests are passing. Thanks!

So the final flow is something like :

...
.splitWhen(_.isLeft)
.prefixAndTail(1)
.map { /*...snip...*/ }
.concatSubstreams
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))

And not

...
.splitWhen(_.isLeft)
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
.prefixAndTail(1)
.map { /*...snip...*/ }
.concatSubstreams

[mdedetrich]

Perfect! Closing ticket

[pjfanning]

@mdedetrich would you have time to review the Pekko 1.1 migration docs? Would it make sense to suggest that the SupervisionStrategy should be be defined after the concatSubstreams call (if any)?

https://pekko.apache.org/docs/pekko/1.1/migration/migration-guide-1.0.x-1.1.x.html

.splitWhen(_.isLeft)
.prefixAndTail(1)
.map { /*...snip...*/ }
.concatSubstreams
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))

[mdedetrich]

Sure, I'll do this today

[mdedetrich]

PR created at #1391