nathanmarz/storm · error · InterruptedException

Disruptor processing interrupted

Error message

Disruptor processing interrupted

What it means

Thrown inside DisruptorQueue.consumeBatchToCursor when the consumer thread waiting on/processing the ring buffer is interrupted (InterruptedException) while handling events or flushing the cache. It signals that the executor's batching thread was interrupted, typically during topology shutdown, and the interrupt is wrapped and rethrown so Storm's shutdown logic can recognize it.

Solutions

1. If seen during topology kill/deactivate, it is expected shutdown behavior and can be ignored.
2. If seen during normal operation, check for external code calling Thread.interrupt() on Storm worker threads.
3. Ensure custom event handlers passed to the queue do not swallow or spuriously trigger interrupts.
4. Restart the worker if the consumer thread is left in a bad state after an unexpected interrupt.

When it happens

Trigger: Thrown at storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java:102 when the library encounters an invalid state.

Common situations: See trigger scenarios.

AI-assisted analysis of nathanmarz/storm@cdb116e942 (2026-09-12). Data as JSON: /api/errors/a0efdc2156db8952. Report an issue: GitHub.

Appendix: source

Thrown at storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java:102

        }
    }
    
    
    private void consumeBatchToCursor(long cursor, EventHandler<Object> handler) {
        for(long curr = _consumer.get() + 1; curr <= cursor; curr++) {
            try {
                MutableObject mo = _buffer.get(curr);
                Object o = mo.o;
                mo.setObject(null);
                if(o==FLUSH_CACHE) {
                    Object c = null;
                    while(true) {                        
                        c = _cache.poll();
                        if(c==null) break;
                        else handler.onEvent(c, curr, true);
                    }
                } else if(o==INTERRUPT) {
                    throw new InterruptedException("Disruptor processing interrupted");
                } else {
                    handler.onEvent(o, curr, curr == cursor);
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        //TODO: only set this if the consumer cursor has changed?
        _consumer.set(cursor);
    }
    
    /*
     * Caches until consumerStarted is called, upon which the cache is flushed to the consumer
     */
    public void publish(Object obj) {
        try {
            publish(obj, true);
        } catch (InsufficientCapacityException ex) {

View on GitHub (pinned to cdb116e942)