{"record":{"id":"6eb5e821f1135023","repo":"apache/pulsar","slug":"queue-is-terminated","errorCode":null,"errorMessage":"Queue is terminated","messagePattern":"Queue is terminated","errorType":"exception","errorClass":"InterruptedException","httpStatus":null,"severity":"error","filePath":"pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java","lineNumber":198,"sourceCode":"        put(e);\n        return true;\n    }\n\n    @Override\n    public boolean offer(T e, long timeout, TimeUnit unit) {\n        // Queue is unbounded and it will never reject new items\n        put(e);\n        return true;\n    }\n\n    @Override\n    public T take() throws InterruptedException {\n        headLock.lockInterruptibly();\n\n        try {\n            while (SIZE_UPDATER.get(this) == 0) {\n                if (terminated) {\n                    throw new InterruptedException(\"Queue is terminated\");\n                }\n                isNotEmpty.await();\n            }\n\n            T item = data[headIndex.value];\n            data[headIndex.value] = null;\n            headIndex.value = (headIndex.value + 1) & (data.length - 1);\n            if (SIZE_UPDATER.decrementAndGet(this) > 0) {\n                // There are still entries to consume\n                isNotEmpty.signal();\n            }\n            return item;\n        } finally {\n            headLock.unlock();\n        }\n    }\n\n    @Override","sourceCodeStart":180,"sourceCodeEnd":216,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java#L180-L216","documentation":"GrowableArrayBlockingQueue.take() blocks waiting for an element, but if the queue has been terminated while empty, it throws InterruptedException('Queue is terminated') instead of blocking forever. Termination is a permanent state of the queue.","triggerScenarios":"Calling take() on a queue whose terminate() has been called while size == 0, or being blocked in take() when another thread terminates the queue and wakes the awaiter.","commonSituations":"Consumer threads still in take() during shutdown after producer called terminate(); not handling InterruptedException in the consumer loop.","solutions":["Handle InterruptedException in the consumer loop and exit cleanly on queue termination","Check isTerminated() before entering take loops","Propagate or restore the interrupt status rather than swallowing it"],"exampleFix":"// before\nwhile (running) {\n    T item = queue.take(); // throws after termination\n    process(item);\n}\n// after\ntry {\n    while (running) {\n        T item = queue.take();\n        process(item);\n    }\n} catch (InterruptedException e) {\n    Thread.currentThread().interrupt(); // queue terminated\n}","handlingStrategy":"try-catch","validationCode":"if (queue.isTerminated()) {\n    return; // don't enter take()\n}","typeGuard":null,"tryCatchPattern":"try {\n    T item = queue.take();\n} catch (InterruptedException e) {\n    Thread.currentThread().interrupt();\n    return; // queue terminated or shutdown requested\n}","preventionTips":["Always wrap take() in try/catch for InterruptedException","Check termination state before consumer loops","Restore interrupt status when catching InterruptedException"],"tags":["java","interrupted-exception","queue-termination"],"backgroundTag":"queue-terminated","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}