File: LogRotatorSink fix completition

[tg44]

issue come up with; https://discuss.lightbend.com/t/writing-element-each-to-its-own-file-problem-on-last-element/7696 also #2553

I need some help, bcs I see something really odd! I will write down the context to get faster to the point;

The logRotator is a custom sink stage. When it gets an element, it runs a function, and if the input function returns with a Some, creates a new subSink, and send the element to that sink.

The current problem appears when the subsink is still initializing (not consuming the dangling data) while the input gets an onUpstreamFinish. At this point the stage completes without giving down the last element to the last sink.

My idea was to simply override the input handler when the rotation happens and we still waiting to the data to be consumed by the new subsink, and when we are going back to the normal operation we complete the stage (if it should be completed).

And I get a timeout exception... So started to paint with printlns the whole code to find out what is actually happening, and I think even if the input handler overrides the onUpstreamFinish somehow the postStop called.

What am I missing? With an overridden onUpstreamFinish the stage should wait until I manually call the completeStage() as I know. How can I debug this further?

Test output:

rotate
s
pull with dangling
normal
rotate
s
f
pull with dangling
normal
rotate
delayedfinish
postStopFinish
s
f

This means that after the rotate, the rotated input handler called with the delayed finish, then somehow the postStop, and after these the new sink starts, and the old finishes. The new sink never finished, so the promise never succeeds.

[johanandren]

Sounds like it could perhaps be because of the auto-close of dangling substreams introduced in Akka 2.6.10 to protect against leaking substreams when the parent stage completes. Couldn't see if that was the case from a quick review of the code though, does it perhaps need to be setup with keepGoing as it is a sink and we want it to continue even though all ports are closed?

[tg44]

That was it! Now it is working!

[tg44]

I removed an unused param in a private class's private subclass. I can revert it instead of the mima exclude. I think it is mergable BTW.

[seglo]

LGTM. Nice re-factor.

I removed an unused param in a private class's private subclass.

I can't think of a reason to keep it and it's internal.

[tg44]

I just fixed the typo, the link validator breaking for some unknown reason. (The refactor mostly needed for the debugging, the bugfix is the key benefit with the added test!)

[seglo]

Thanks @tg44

[reppners]

I'm using Akka 2.6.10 and Alpakka 2.0.2 and stumbled over this issue. Nice to see it is fixed already! Is there a timeline when a release containing this PR will be available or if this may be backported to 2.x?

[seglo]

@reppners Their should be timestamped snapshots already available with this fix (they're published each master build). If you would like to try it immediately I would suggest referencing a snapshot. Otherwise, we intend to release a 3.0.0 milestone in the next few weeks, but I'm not sure a timeline for a final 3.0.0.