apache/pulsar · error · InterruptedException

Queue is terminated

Error message

Queue is terminated

What it means

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.

Source

Thrown at pulsar-common/src/main/java/org/apache/pulsar/common/util/collections/GrowableArrayBlockingQueue.java:198

        put(e);
        return true;
    }

    @Override
    public boolean offer(T e, long timeout, TimeUnit unit) {
        // Queue is unbounded and it will never reject new items
        put(e);
        return true;
    }

    @Override
    public T take() throws InterruptedException {
        headLock.lockInterruptibly();

        try {
            while (SIZE_UPDATER.get(this) == 0) {
                if (terminated) {
                    throw new InterruptedException("Queue is terminated");
                }
                isNotEmpty.await();
            }

            T item = data[headIndex.value];
            data[headIndex.value] = null;
            headIndex.value = (headIndex.value + 1) & (data.length - 1);
            if (SIZE_UPDATER.decrementAndGet(this) > 0) {
                // There are still entries to consume
                isNotEmpty.signal();
            }
            return item;
        } finally {
            headLock.unlock();
        }
    }

    @Override

View on GitHub (pinned to 820761864e)

Solutions

1. Handle InterruptedException in the consumer loop and exit cleanly on queue termination
2. Check isTerminated() before entering take loops
3. Propagate or restore the interrupt status rather than swallowing it

Example fix

// before
while (running) {
    T item = queue.take(); // throws after termination
    process(item);
}
// after
try {
    while (running) {
        T item = queue.take();
        process(item);
    }
} catch (InterruptedException e) {
    Thread.currentThread().interrupt(); // queue terminated
}

if (queue.isTerminated()) {
    return; // don't enter take()
}

try {
    T item = queue.take();
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    return; // queue terminated or shutdown requested
}

When it happens

Trigger: 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.

Common situations: Consumer threads still in take() during shutdown after producer called terminate(); not handling InterruptedException in the consumer loop.

Related errors

• ${key} already exists in the dynamicConfigurationMap
• Topic factory failed to create topic
• No more range can assigned to new consumer, assigned consume
• Range conflict with consumer ${conflictingConsumer}
• Error creating client for HealthChecker

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/6eb5e821f1135023. Report an issue: GitHub.