{"record":{"id":"29a52181ccbb278d","repo":"apache/dolphinscheduler","slug":"all-exclusivethreadtaskexecutorworker-are-busy","errorCode":null,"errorMessage":"All ExclusiveThreadTaskExecutorWorker are busy","messagePattern":"All ExclusiveThreadTaskExecutorWorker are busy","errorType":"exception","errorClass":"TaskExecutorRuntimeException","httpStatus":null,"severity":"error","filePath":"dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/container/AbstractTaskExecutorContainer.java","lineNumber":63,"sourceCode":"    protected final TaskExecutorWorkers taskExecutorWorkers;\n\n    public AbstractTaskExecutorContainer(final TaskExecutorContainerConfig containerConfig) {\n        final String threadPoolFormat = containerConfig.getContainerName() + \"-worker-%d\";\n        final int threadPoolSize = containerConfig.getTaskExecutorThreadPoolSize();\n        this.taskExecutorThreadPool = ThreadUtils.newDaemonFixedThreadExecutor(threadPoolFormat, threadPoolSize);\n        this.taskExecutorWorkers = new TaskExecutorWorkers(threadPoolSize);\n        this.taskExecutorAssignmentTable = new TaskExecutorAssignmentTable();\n        startAllThreadTaskExecutorWorker();\n    }\n\n    @Override\n    public void dispatch(final ITaskExecutor taskExecutor) {\n        synchronized (this) {\n            Optional<TaskExecutorWorker> taskExecutorWorkerCandidate = getTaskExecutorWorkerCandidate(taskExecutor);\n            if (!taskExecutorWorkerCandidate.isPresent()) {\n                log.info(\"All ExclusiveThreadTaskExecutorWorker are busy, cannot submit taskExecutor(id={})\",\n                        taskExecutor.getId());\n                throw new TaskExecutorRuntimeException(\"All ExclusiveThreadTaskExecutorWorker are busy\");\n            }\n            final TaskExecutorWorker taskExecutorWorker = taskExecutorWorkerCandidate.get();\n            taskExecutorWorker.registerTaskExecutor(taskExecutor);\n            taskExecutorAssignmentTable.registerTaskExecutor(taskExecutor, taskExecutorWorker);\n        }\n    }\n\n    @Override\n    public void start(final ITaskExecutor taskExecutor) {\n        final Integer workerId = taskExecutorAssignmentTable.getTaskExecutorWorkerId(taskExecutor);\n        if (workerId == null) {\n            throw new IllegalStateException(\n                    \"The taskExecutor: \" + taskExecutor.getId() + \" is not registered to any worker\");\n        }\n        final TaskExecutorWorker taskExecutorWorker = taskExecutorWorkers.getWorkerById(workerId);\n        taskExecutorWorker.fireTaskExecutor(taskExecutor);\n    }\n","sourceCodeStart":45,"sourceCodeEnd":81,"githubUrl":"https://github.com/apache/dolphinscheduler/blob/02eac45a1b6676e639fcbfb4be2243de5771b05d/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/container/AbstractTaskExecutorContainer.java#L45-L81","documentation":"AbstractTaskExecutorContainer.dispatch assigns each task executor to an exclusive per-thread worker. When getTaskExecutorWorkerCandidate finds no free worker (all worker threads are occupied by other task executors), dispatch throws TaskExecutorRuntimeException instead of queueing, providing backpressure to the submitter.","triggerScenarios":"Submitting more concurrent task executors to the container than there are worker threads: dispatch() called while every ExclusiveThreadTaskExecutorWorker already has a registered executor.","commonSituations":"Burst of task instances exceeding workerThreadNum/parallelism configured for the container; long-running tasks hogging all workers; under-provisioned master with high workflow concurrency.","solutions":["Increase the container's worker thread count / parallelism configuration","Reduce concurrent workflow/task submission or add queueing before dispatch","Retry dispatch after a delay (or monitor worker availability before submitting)","Scale out masters to spread the load"],"exampleFix":"// before\ntaskExecutorContainer.dispatch(taskExecutor); // may throw when busy\n// after\ntry {\n    taskExecutorContainer.dispatch(taskExecutor);\n} catch (TaskExecutorRuntimeException e) {\n    // requeue and retry later\n    pendingExecutors.offer(taskExecutor);\n}","handlingStrategy":"fallback","validationCode":"// Throttle submissions to the container's worker count:\nif (activeDispatched >= configuredWorkerThreads) {\n    pendingExecutors.offer(taskExecutor); // queue instead of dispatching now\n    return;\n}","typeGuard":null,"tryCatchPattern":"try {\n    container.dispatch(taskExecutor);\n} catch (TaskExecutorRuntimeException e) {\n    if (e.getMessage().contains(\"are busy\")) {\n        pendingExecutors.offer(taskExecutor); // retry with backoff\n    } else throw e;\n}","preventionTips":["Size worker threads >= expected concurrent task executors","Implement an application-level queue in front of dispatch","Alert on container saturation; scale masters when dispatch failures occur"],"tags":["task-executor","capacity","concurrency","backpressure"],"backgroundTag":"resource-exhausted","analyzedSha":"02eac45a1b6676e639fcbfb4be2243de5771b05d","analyzedAt":"2026-09-06T17:43:00.555Z","contentChangedAt":"2026-09-06T17:43:00.555Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}