apache/flink · error · RejectedExecutionException

The ExecutorService is shut down already. No Callables can b

Error message

The ExecutorService is shut down already. No Callables can be executed.

What it means

DirectExecutorService can be constructed in a mode (triggerRejectedExecutionException) where any submission after shutdown() throws RejectedExecutionException with this message. This matches the standard ExecutorService contract that a shut-down executor refuses new work, letting callers detect use-after-shutdown deterministically instead of silently running the task.

Source

Thrown at flink-core/src/main/java/org/apache/flink/util/concurrent/DirectExecutorService.java:252

        }

        if (iterator.hasNext()) {
            throw new TimeoutException("Could not finish execution of tasks within time.");
        } else {
            throw new ExecutionException("No tasks finished successfully.", exception);
        }
    }

    @Override
    public void execute(@Nonnull Runnable command) {
        throwRejectedExecutionExceptionIfShutdown();

        command.run();
    }

    private void throwRejectedExecutionExceptionIfShutdown() {
        if (isShutdown() && triggerRejectedExecutionException) {
            throw new RejectedExecutionException(
                    "The ExecutorService is shut down already. No Callables can be executed.");
        }
    }

    static class CompletedFuture<V> implements Future<V> {
        private final V value;
        private final Exception exception;

        CompletedFuture(V value, Exception exception) {
            this.value = value;
            this.exception = exception;
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            return false;
        }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Fix ordering so no submissions happen after shutdown(); cancel the producers first, then shut down.
2. Guard submission sites with !executor.isShutdown() when late tasks are legal and should be dropped.
3. Create a fresh DirectExecutorService per lifecycle scope instead of reusing one across shutdown boundaries.

Example fix

// before
executor.shutdown();
executor.execute(task);

// after
executor.shutdown();
if (!executor.isShutdown()) { executor.execute(task); } // or reorder shutdown after all submissions

if (!executor.isShutdown()) { executor.execute(task); }

try { executor.execute(task); }
catch (RejectedExecutionException e) { /* executor shut down: drop or re-route the task */ }

When it happens

Trigger: Calling execute(), submit(), invokeAll(), or invokeAny() on a DirectExecutorService after shutdown(), when the service was created with rejection-on-shutdown enabled (the default constructor enables it).

Common situations: Components sharing a DirectExecutorService instance that outlives a shutdown in tests or teardown; lifecycle races where a callback submits work during close(); reusing a static executor across test cases that shuts it down per test.

Related errors

• Resource guard was already closed.
• Could not close resource.
• Task has been cancelled.
• No tasks finished successfully.
• Could not finish execution of tasks within time.

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