{"record":{"id":"a0efdc2156db8952","repo":"nathanmarz/storm","slug":"disruptor-processing-interrupted","errorCode":null,"errorMessage":"Disruptor processing interrupted","messagePattern":"Disruptor processing interrupted","errorType":"exception","errorClass":"InterruptedException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java","lineNumber":102,"sourceCode":"        }\n    }\n    \n    \n    private void consumeBatchToCursor(long cursor, EventHandler<Object> handler) {\n        for(long curr = _consumer.get() + 1; curr <= cursor; curr++) {\n            try {\n                MutableObject mo = _buffer.get(curr);\n                Object o = mo.o;\n                mo.setObject(null);\n                if(o==FLUSH_CACHE) {\n                    Object c = null;\n                    while(true) {                        \n                        c = _cache.poll();\n                        if(c==null) break;\n                        else handler.onEvent(c, curr, true);\n                    }\n                } else if(o==INTERRUPT) {\n                    throw new InterruptedException(\"Disruptor processing interrupted\");\n                } else {\n                    handler.onEvent(o, curr, curr == cursor);\n                }\n            } catch (Exception e) {\n                throw new RuntimeException(e);\n            }\n        }\n        //TODO: only set this if the consumer cursor has changed?\n        _consumer.set(cursor);\n    }\n    \n    /*\n     * Caches until consumerStarted is called, upon which the cache is flushed to the consumer\n     */\n    public void publish(Object obj) {\n        try {\n            publish(obj, true);\n        } catch (InsufficientCapacityException ex) {","sourceCodeStart":84,"sourceCodeEnd":120,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java#L84-L120","documentation":"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.","triggerScenarios":"Thrown at storm-core/src/jvm/backtype/storm/utils/DisruptorQueue.java:102 when the library encounters an invalid state.","commonSituations":"See trigger scenarios.","solutions":["If seen during topology kill/deactivate, it is expected shutdown behavior and can be ignored.","If seen during normal operation, check for external code calling Thread.interrupt() on Storm worker threads.","Ensure custom event handlers passed to the queue do not swallow or spuriously trigger interrupts.","Restart the worker if the consumer thread is left in a bad state after an unexpected interrupt."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":null,"preventionTips":[],"tags":[],"backgroundTag":null,"analyzedSha":"cdb116e942666973bc4eaa0df098d5bab82739e7","analyzedAt":"2026-09-12T14:30:00.714Z","contentChangedAt":"2026-09-12T14:30:00.714Z","schemaVersion":2},"datasetVersion":"2026-09-16T09:17:16.951Z"}