STORM-342: Message loss, executor hang, or message disorder due to contention in Disruptor queue under multi-thread mode.

STORM-342: Message loss, executor hang, or message disorder
-------------------------

Disruptor helper class contains a potential contention bug between consumer and producer. It can cause consume queue hang, message loss, or message disorder.

```java
Disruptor.java
class Disruptor {
...
    public void publish(Object obj, boolean block) throws InsufficientCapacityException {
        if(consumerStartedFlag) {
            final long id;
            if(block) {
                id = _buffer.next();
            } else {
                id = _buffer.tryNext(1);
            }
            final MutableObject m = _buffer.get(id);
            m.setObject(obj);
            _buffer.publish(id);
        } else {
            _cache.add(obj);
            if(consumerStartedFlag) flushCache();
        }
    }
    
    public void consumerStarted() {
        if(!consumerStartedFlag) {
            consumerStartedFlag = true;
            flushCache();
        }
    }
}
```

Consumer
```lisp
;;Executor thead
  (disruptor/consumer-started! receive-queue)
  (fn []            
     (disruptor/consume-batch-when-available receive-queue event-handler)
```

Howto: Executor Hang, message loss:
------------------------
1. [Consumer Thread] consumer not started.
2. [Producer A Thread] publish message "1", as "consumerStartedFlag" == false, it will be added it into cache.
3. [Consumer Thread] consumerStarted() is called. consumerStartedFlag is set to true, but flushCache() is not called yet.
4. As "consumerStartedFlag" is true now, new produced message will be published to RingBuffer.
5. [Producer B Thread] generates enough message, and make RingBuffer full.
6. [Consumer Thread] flushCache() is called in consumerStarted() 
7. [Consumer Thread] FLUSH_CACHE object is published RingBuffer in blocking way, As now RingBuffer is full, the consumer thread will be blocked.
8. [Consumer Thread] consumeBatch() will never called, so the RingBuffer is always full, and the consumer thread is always blocked.

Howto: Message Disorder
-----------------------------------
1. [Consumer Thread] consumer not started.
2. [Producer A Thread] publish message "1", as "consumerStartedFlag" == false, it will be added it into cache.
3. [Consumer Thread] consumerStarted() is called. consumerStartedFlag is set to true, but flushCache() is not called yet.
4. As "consumerStartedFlag" is true now, new produced message will be published to RingBuffer.
5. [Producer A Thread] publish a new message "2", it will be published directly in RingBuffer.
6. [Consumer Thread] flushCache() is called in consumerStarted() 
7. [Consumer Thread] FLUSH_CACHE message is published RingBuffer, FLUSH_CACHE message is written after message "2".
8. [Consumer Thread] consumeBatch() is called, first it picks "2", then it picks FLUSH_CACHE, will represents "1"
9. We produce in Producer A Thread in order "1", "2", but we received in consumer thread "2", "1"
10. Message order is wrong.
Can we include e as the cause of the RuntimeException.  There have been times where I thought code should be unreachable, and when we did reach it in production I had no way to debug what happened.