{"record":{"id":"947b61c601bc13f6","repo":"apache/pulsar","slug":"worker-workerid-was-not-yet-removed-after-a-pri","errorCode":null,"errorMessage":"Worker ${workerId} was not yet removed after a prior drain op; try later","messagePattern":"Worker (.+?) was not yet removed after a prior drain op; try later","errorType":"http","errorClass":"SchedulerManager.WorkerNotRemovedAfterPriorDrainException","httpStatus":412,"severity":"error","filePath":"pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java","lineNumber":283,"sourceCode":"    public Future<?> drainIfNotInProgress(String workerId) {\n        if (drainInProgressFlag.compareAndSet(false, true)) {\n            try {\n                Set<String> availableWorkers = getCurrentAvailableWorkers();\n                if (availableWorkers.size() <= 1) {\n                    throw new TooFewWorkersException();\n                }\n\n                // A worker must be specified at this point. This would be set up by the caller.\n                Objects.requireNonNull(workerId);\n\n                // [We can get stricter, and require that every drain op be followed up with a cleanup of the\n                // corresponding worker before any other drain op, so that the drainOpStatusMap should be empty\n                // at the next drain operation.]\n                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) {","sourceCodeStart":265,"sourceCodeEnd":301,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java#L265-L301","documentation":"SchedulerManager tracks completed/in-progress drain operations in drainOpStatusMap; if a previous drain of the same workerId has not been cleaned up yet, a new drain of that worker throws WorkerNotRemovedAfterPriorDrainException. The worker from the prior drain is still present in the cluster view, so the manager refuses to start another drain of it.","triggerScenarios":"Calling drain twice for the same workerId before the first drain fully completed and the worker was removed from the cluster; the drained worker is still registered (has not actually shut down) so cleanup of drainOpStatusMap never happens.","commonSituations":"Scripts that re-issue a drain command after seeing partial failure; the drained worker process failed to terminate so its registration lingers; rapid retry of drain before the worker de-registers from the metadata store.","solutions":["Wait for the worker to actually leave the cluster (verify it is gone from the worker list) before draining again.","Check the drain status via getDrainStatus(workerId) and only re-drain once it reports completion and the worker is removed.","Force-verify the drained worker process is stopped and de-registered from the metadata store; then retry."],"exampleFix":"// before\nworker.drainIfNotInProgress(workerId); // re-drain too early\n// after\nLongRunningProcessStatus st = worker.getDrainStatus(workerId);\nif (st.getStatus() != LongRunningProcessStatus.Status.SUCCESS) {\n    log.info(\"Prior drain not finished; retry later\");\n    return;\n}\nworker.drainIfNotInProgress(workerId);","handlingStrategy":"retry","validationCode":"// check prior drain is finished and worker removed before re-draining\nLongRunningProcessStatus prev = worker.getDrainStatus(workerId);\nboolean alreadyDraining = prev != null\n    && prev.getStatus() == LongRunningProcessStatus.Status.RUNNING;\nif (alreadyDraining) {\n    throw new IllegalStateException(\"Wait for prior drain of \" + workerId + \" to finish\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    worker.drainIfNotInProgress(workerId);\n} catch (WorkerNotRemovedAfterPriorDrainException e) {\n    // poll getDrainStatus until previous drain completes, then retry\n}","preventionTips":["Confirm the worker actually shut down and deregistered before re-draining","Poll getDrainStatus between drain attempts","Fix workers that fail to terminate so registration does not linger"],"tags":["pulsar-functions","drain","lifecycle"],"backgroundTag":"prior-drain-not-cleaned-up","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}