apache/flink · error · FlinkRuntimeException

exception in queue future completion

Error message

exception in queue future completion

What it means

FutureCompletingBlockingQueue#take waits on the queue's availability future. If that future completed exceptionally (ExecutionException / CompletionException) the code re-throws as a FlinkRuntimeException, noting 'this should never happen'. The availability future is an internal signaling primitive that is not expected to complete with an error.

Source

Thrown at flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueue.java:235

     * <p>Get and remove the first element from the queue. The call blocks if the queue is empty.
     * The problem with this method is that it may loop internally until an element is available and
     * that way eagerly reset the availability future. If a consumer thread is blocked in taking an
     * element, it will receive availability notifications from {@link #notifyAvailable()} and
     * immediately reset them by calling {@link #poll()} and finding the queue empty.
     *
     * @return the first element in the queue.
     * @throws InterruptedException when the thread is interrupted.
     */
    @VisibleForTesting
    public T take() throws InterruptedException {
        T next;
        while ((next = poll()) == null) {
            // use the future to wait for availability to avoid busy waiting
            try {
                getAvailabilityFuture().get();
            } catch (ExecutionException | CompletionException e) {
                // this should never happen, but we propagate just in case
                throw new FlinkRuntimeException("exception in queue future completion", e);
            }
        }
        return next;
    }

    /**
     * Get and remove the first element from the queue. Null is returned if the queue is empty. If
     * this makes the queue empty (takes the last element) or finds the queue already empty, then
     * this resets the availability notifications. The next call to {@link #getAvailabilityFuture()}
     * will then return a non-complete future that completes only the next time that the queue
     * becomes non-empty or the {@link #notifyAvailable()} method is called.
     *
     * @return the first element from the queue, or Null if the queue is empty.
     */
    public T poll() {
        lock.lock();
        try {
            if (queue.size() == 0) {

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Capture the wrapped cause to identify what completed the availability future exceptionally.
2. Ensure no custom code touches the FutureCompletingBlockingQueue's availability future via completeExceptionally.
3. If reproducible, file a Flink JIRA with the stack trace; this path is documented as unreachable.
4. Restart the job; transient occurrences during teardown are not actionable.

try {
    queue.take();
} catch (FlinkRuntimeException e) {
    if (e.getMessage().contains("queue future completion")) {
        // internal invariant violated; restart the operator
        throw e;
    }
    throw e;
}

When it happens

Trigger: The internal availability future completed exceptionally, indicating a programming error in how the queue's availability is signaled (e.g., a buggy custom component calling completeExceptionally on the future).

Common situations: Essentially a bug path; would indicate framework-internal corruption of the queue's signaling future. Not triggered by normal user data or config.

Related errors

• URL is invalid. This should not happen.
• Corrupt artifact fetching state.
• Source fetch execution was interrupted
• SplitFetcher thread %d received unexpected exception while p
• The thread was interrupted while waiting for a fetcher task.

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/a3f11c2efceb3af1. Report an issue: GitHub.