{"record":{"id":"1b1bad5b9bb6bf8c","repo":"apache/pulsar","slug":"another-drain-is-in-progress","errorCode":null,"errorMessage":"Another drain is in progress","messagePattern":"Another drain is in progress","errorType":"http","errorClass":"SchedulerManager.DrainInProgressException","httpStatus":409,"severity":"warning","filePath":"pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java","lineNumber":297,"sourceCode":"                if (drainOpStatusMap.containsKey(workerId)) {\n                    String warnString = \"Worker \" + workerId\n                            + \" was not removed yet from SchedulerManager after previous drain op\";\n                    log.warn(warnString);\n                    throw new WorkerNotRemovedAfterPriorDrainException();\n                }\n\n                if (!availableWorkers.contains(workerId)) {\n                    log.info().attr(\"workerId\", workerId)\n                            .log(\"invokeDrain was called for a worker which is not currently active\");\n                    throw new UnknownWorkerException();\n                }\n\n                return drain(workerId);\n            } finally {\n                drainInProgressFlag.set(false);\n            }\n        } else {\n            throw new DrainInProgressException();\n        }\n    }\n\n    public LongRunningProcessStatus getDrainStatus(String workerId) {\n        long startTime = System.nanoTime();\n        LongRunningProcessStatus status = Optional.ofNullable(workerId).map(id ->\n                Optional.ofNullable(drainOpStatusMap.get(id)).map(opStatus ->\n                        switch (opStatus) {\n                            case DrainCompleted ->\n                                    LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.SUCCESS);\n                            case DrainInProgress ->\n                                    LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.RUNNING);\n                            case DrainNotInProgress ->\n                                    LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.NOT_RUN);\n                        }).orElse(\n                        LongRunningProcessStatus.forError(\"Worker \" + id + \" not found in drain records\")\n                )\n        ).orElse(","sourceCodeStart":279,"sourceCodeEnd":315,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java#L279-L315","documentation":"SchedulerManager serializes drain operations with an atomic drainInProgressFlag; if another drain is already running, a concurrent call throws DrainInProgressException ('Another drain is in progress'). Only one drain may run at a time cluster-wide per this manager.","triggerScenarios":"Issuing drain for a second worker while the first drain is still executing; parallel scripts draining several workers simultaneously; retrying a drain before the previous one finished.","commonSituations":"Automated rolling shutdown that drains multiple workers in parallel instead of sequentially; slow drain (large assignment transfer) overlapping with the next drain request.","solutions":["Run drain operations sequentially: wait for each drain to complete (poll getDrainStatus) before starting the next.","Catch DrainInProgressException and retry with backoff.","Refactor automation to a single-threaded drain queue."],"exampleFix":"// before\nworkers.forEach(w -> worker.drainIfNotInProgress(w)); // parallel drains\n// after\nfor (String w : workers) {\n    worker.drainIfNotInProgress(w);\n    while (worker.getDrainStatus(w).getStatus() == LongRunningProcessStatus.Status.RUNNING) {\n        Thread.sleep(1000);\n    }\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n    worker.drainIfNotInProgress(workerId);\n} catch (DrainInProgressException e) {\n    // another drain is running: wait, then retry with backoff\n}","preventionTips":["Drain one worker at a time sequentially","Wait for getDrainStatus to report completion before the next drain","Use a single coordinator/queue for drain automation"],"tags":["pulsar-functions","drain","concurrency"],"backgroundTag":"operation-already-in-progress","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}