Put all Collector actions in one execute

[johnynek]

No description provided.

[johnynek]

@singhala @jnievelt guys

Review help would be great here.

This is a non-trivial change, and hopefully BIG improvement in storm correctness and error logging

[ryanlecompte]

Was your intention for this to block if queue is an ArrayBlockingQueue with a fixed size? What actually will happen is that a java.lang.IllegalStateException: Queue full exception will be thrown instead of it actually blocking (you would need to call put instead for the blocking behavior).

[johnynek]

Nice catch. Fixed.

[ryanlecompte]

Looks nice!

[ianoc]

If its an always, why not bake the behavior into these rather than at touch points?

[ianoc]

Could we supply a more scala idiomatic one, return Option[T] and internally do the decrement if not null?

[johnynek]

I am trying to make it easy to get that right, but if we require every subclass to call decrement, some will get it wrong. By contrast, if we do it below, we can audit it and test it once. This method is just for the adapter to existing java classes.

Option I'll do. That was we being OCD about allocations.... bad move.

[ianoc]

So with the the atomic thing, i was thinking it might be more have the downstream api but then final functions which call them and do the decrement/increment. Such that there is only 2 places we need to ensure correctness. Not a big thing, its only a tiny class anyway.

[ianoc]

Looks great, nice work. All comments are really very minor