{"record":{"id":"a3f11c2efceb3af1","repo":"apache/flink","slug":"exception-in-queue-future-completion","errorCode":null,"errorMessage":"exception in queue future completion","messagePattern":"exception in queue future completion","errorType":"exception","errorClass":"FlinkRuntimeException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueue.java","lineNumber":235,"sourceCode":"     * <p>Get and remove the first element from the queue. The call blocks if the queue is empty.\n     * The problem with this method is that it may loop internally until an element is available and\n     * that way eagerly reset the availability future. If a consumer thread is blocked in taking an\n     * element, it will receive availability notifications from {@link #notifyAvailable()} and\n     * immediately reset them by calling {@link #poll()} and finding the queue empty.\n     *\n     * @return the first element in the queue.\n     * @throws InterruptedException when the thread is interrupted.\n     */\n    @VisibleForTesting\n    public T take() throws InterruptedException {\n        T next;\n        while ((next = poll()) == null) {\n            // use the future to wait for availability to avoid busy waiting\n            try {\n                getAvailabilityFuture().get();\n            } catch (ExecutionException | CompletionException e) {\n                // this should never happen, but we propagate just in case\n                throw new FlinkRuntimeException(\"exception in queue future completion\", e);\n            }\n        }\n        return next;\n    }\n\n    /**\n     * Get and remove the first element from the queue. Null is returned if the queue is empty. If\n     * this makes the queue empty (takes the last element) or finds the queue already empty, then\n     * this resets the availability notifications. The next call to {@link #getAvailabilityFuture()}\n     * will then return a non-complete future that completes only the next time that the queue\n     * becomes non-empty or the {@link #notifyAvailable()} method is called.\n     *\n     * @return the first element from the queue, or Null if the queue is empty.\n     */\n    public T poll() {\n        lock.lock();\n        try {\n            if (queue.size() == 0) {","sourceCodeStart":217,"sourceCodeEnd":253,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/synchronization/FutureCompletingBlockingQueue.java#L217-L253","documentation":"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.","triggerScenarios":"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).","commonSituations":"Essentially a bug path; would indicate framework-internal corruption of the queue's signaling future. Not triggered by normal user data or config.","solutions":["Capture the wrapped cause to identify what completed the availability future exceptionally.","Ensure no custom code touches the FutureCompletingBlockingQueue's availability future via completeExceptionally.","If reproducible, file a Flink JIRA with the stack trace; this path is documented as unreachable.","Restart the job; transient occurrences during teardown are not actionable."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    queue.take();\n} catch (FlinkRuntimeException e) {\n    if (e.getMessage().contains(\"queue future completion\")) {\n        // internal invariant violated; restart the operator\n        throw e;\n    }\n    throw e;\n}","preventionTips":["Never call completeExceptionally on the queue's availability future.","Treat occurrences as framework bugs; capture the cause and file a JIRA.","Do not subclass FutureCompletingBlockingQueue to add custom signaling."],"tags":["blocking-queue","future","internal","flink-source"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}