STORM-2039: backpressure refactoring in worker and executor

[abellina]

For 1.x branch, will submit a PR for 2.x branch.

@zhuoliu worked on the initial refactor. I added some due code review comments.

[knusbaum]

This feels like the wrong place for this.
For some reason the backpressure is tacked onto the credentials timer?

I know this was already the way it worked, but maybe there's a better place for it now? I don't think it's good that the value of the task.credentials.poll.secs config affects the frequency of backpressure polling.

[abellina]

yeah, agreed. I'll add another timer for this.

[knusbaum]

+1 for the changes. My comments are suggestions for a couple more.
The travis failures look unrelated. Storm Core passes.

[abellina]

@knusbaum I made the changes suggested. Also removed an extra fn wrapper on that function triggered by the timer.

[knusbaum]

+1

[zhuoliu]

@abellina looks good to me, just one extra suggestion:
In
https://github.com/apache/storm/blob/master/storm-core/src/clj/org/apache/storm/daemon/worker.clj#L156
To reduce the traffic to ZK, we checked the previous flag to update to Zookeeper only when it is necessary (the flag has changed), but still there may some weird case we have inconsistency between local "prev-backpressure-flag" and the worker backpressure node at Zookeeper (see STORM-1949).
(One such weird case I would think of is: the worker-backpressure call does not update the zk flag successfully but no exception was caught and the local flag updating still succeeds)

So to strike a tradeoff and avoid the possible wrong stall of a topology, could you add do a random number generation here, e.g., generate a number between 0-9, if it is zero, we will call the try block no matter "prev-backpressure-flag" and "curr-backpressure-flag" is equal or not.
This will make sure the local flag and remote flag is synchronized and the ZK traffic would be once per second per worker. (the cyclic function is called every 100ms). Or one every 5-10 second is still acceptable since this is only needed rarely.
Using (currentTimeStampMs / 1000 % intervalInSeconds == 0) as the update condition might make more sense than using random number. For intervalInSeconds, we can use 5 or 10. Most of the ZK traffic would be read. Assume ZK (3 nodes) can sustain ~50K req/second. 5% of its capacity can satisfy 25K workers if setting the intervalInSeconds as 10.

Thanks a lot!

[abellina]

@zhuoliu thanks for the comment. I am taking a look at STORM-1949 to try and reproduce it, and will comment there. I think that is a separate issue from this PR so we should see a potential fix to STORM-1949 in a different PR. Are you ok with this going in as is?

[zhuoliu]

sure, Alex. Really appreciate if you can submit a separate PR for 1949.
+1 to this one!

[zhuoliu]

Since with the refactoring we directly access the disruptor queue to get the backpressure status rather than having an additional flag in executor data which is error-prone when flipping (which is very possibly the root cause of the halt in STORM-1949), we may NOT need to add the ADDITIONAL ZK pressure which I mentioned in the above.

[HeartSaVioR]

+1