Comments (9)
@raboof do you know which change broke it?
Any chance reference.conf
settings were overridden?
https://github.com/akka/akka-stream-contrib/blob/master/contrib/src/main/resources/reference.conf
If configured time isn't enough, stage will fail with an IllegalStateException
instead of the expected exception.
(test took 23ms, so config/killSwitch change is the only guesses I have right now as to why it failed...)
P.S.
We're running with unfoldFlow in production for several months now:
https://github.com/thomsonreuters/CM-Well
and no failures in it so far...
Will try to see if I can reproduce it later.
from akka-stream-contrib.
Well, only reproducible in 2.5(.3), and not deterministically.
Fail on both on
& off
fusing (which doesn't matter anyway, since 2.5.X has fusing always enabled, and setting is deprecated).
When all other tests ignored, passes for both on
& off
.
from akka-stream-contrib.
Added some debug prints, and got the following findings:
test code:
val (killSwitch, sink) = source.toMat(TestSink.probe)(Keep.both).run()
val kill = new Exception("KILL!")
sink.ensureSubscription()
killSwitch.abort(kill)
sink.request(1)
sink.expectError(kill)
expectations:
killSwitch invoked, causing a race condition between upstream & downstream cancellations:
scenario 1:
killSwitch invoked -> flow propagates termination-> InHandler.onUpstreamFailure is invoked and stream terminates properly.
scenario 2:
killSwitch invoked -> feedback's OutHandler.onDownstreamFinish -> set scheduler to configured 1 minute to allow scenario 1 to finish properly
in any case, stream should have been canceled.
what actually happened was:
1- killSwitch invoked
2- test request(1) -> output's onPull -> feedback's onPull -> onPush -> push
3- feedback's OutHandler.onDownstreamFinish is invoked
4- InHandler's onUpstreamFailure is invoked
The problem is that killSwitch is still operational after being aborted.
This seems like a bug in KillSwitch!
After being aborted, it shouldn't have handled both onPull
& onPush
.
from akka-stream-contrib.
abort()
is not synchronous, so you cannot be sure the command to abort passes the asynchronous boundary before the request and emit happens. That's a race that is hard to workaround I'm afraid. Only thing I can think of making it less racy is the sleep before requesting.
from akka-stream-contrib.
@johanandren so is this test even needed? perhaps we should just drop the test entirely...
from akka-stream-contrib.
Sounds like should fail instantly when aborted
should cover that it works, but I didn't look into the code.
I'm not quite sure why there needs to be special tests for killswitch here, it doesn't do anything a cancelled/failed/completed stream does not already do.
from akka-stream-contrib.
Since the unfold source closes over a given unknown flow, we need to take careful care of flow termination.
We closes over the given flow from both upstream and downstream.
So, if we get onDownstreamFinish()
being called, we can't (yet) know how to terminate.
So we wait for a configurable amount of time, and wait for either onUpstreamFinish()
or onUpstreamFailure(ex)
.
If configured time exceeded, we terminate with fail(IllegalStateException)
.
Because of that, there's difference between KillSwitch being appended or prepended to flow. (maybe a test holding downstream fail explicitly should be added).
Also, the reason this test was added, was to check the following scenario:
- Inner flow send termination as downstream (
onDownstreamFinish()
is called) - We wait (1m by default) to get flow termination as
onUpstreamFinish()
oronUpstreamFailure(ex)
- Source is being pulled
- We get
onUpstreamFailure(ex)
It's a delicate edge case, and IIRC, I had a bug, pull changed the state in a way such that flow didn't terminate.
from akka-stream-contrib.
Can you perhaps put two probes in there instead, by using Flow.fromSinkAndSource
and have full control over what goes on in the inner flow?
from akka-stream-contrib.
Good idea. Will try to get to it soon and open a quick PR.
from akka-stream-contrib.
Related Issues (20)
- Make the valve state inspection always complete
- Alternative way of defining Graph Stages HOT 3
- Release 0.9 HOT 1
- IntervalBasedRateLimiter should limit rate of messages when frequency is very high (50 000 elements per 1 000ms) failed: 4 was not equal to 3
- FileTailSource and DirectoryChanges have moved to Alpakka HOT 1
- replace scala.Either in PartitionWith Java API HOT 3
- Any interest in integrating actor into a Flow though a GraphStage? HOT 3
- Add maxNumber and maxTime to AccumulateWhileUnchanged
- Publish artifacts for scala 2.13 HOT 2
- Publishing from Travis fails
- Remove unrelated tools
- PassThroughFlow limits the rate of the inner processingFlow HOT 5
- Remove PassThroughFlow which is not superseded by FlowWithContext
- Introduce Token Throttle HOT 3
- TokenThrottle can fail to complete and doesn't consistently pre-fetch tokens
- Publish API documentation to doc.akka.io HOT 7
- Support for sorting fan-in shape with more than two upstreams
- Migrate from travis.org to travis.com HOT 1
- Shared valve switch HOT 1
- Why no more releases? HOT 1
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from akka-stream-contrib.