{"record":{"id":"35caf95752e4a76b","repo":"apache/pulsar","slug":"worker-workerid-is-not-among-the-current-worker","errorCode":null,"errorMessage":"Worker ${workerId} is not among the current workers in the system","messagePattern":"Worker (.+?) is not among the current workers in the system","errorType":"http","errorClass":"SchedulerManager.UnknownWorkerException","httpStatus":400,"severity":"error","filePath":"pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java","lineNumber":289,"sourceCode":"                }\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) {\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);","sourceCodeStart":271,"sourceCodeEnd":307,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java#L271-L307","documentation":"drainIfNotInProgress(workerId) checks the set of currently available workers; if the requested workerId is not among them it throws UnknownWorkerException ('Worker <id> is not among the current workers in the system'). You asked to drain a worker the cluster does not currently see as active.","triggerScenarios":"Draining a workerId that was already removed/never existed; typo in workerId; the target worker crashed or lost metadata-store registration before the drain call; calling drain after the worker already deregistered.","commonSituations":"Stale configuration or scripts referencing a decommissioned worker; retrying a drain after the worker already went away; hostname/worker-id mismatch (using hostname vs configured worker id).","solutions":["List current workers (worker stats / REST) and use an exact existing workerId.","Verify the worker's configured workerId matches the value you pass.","If the worker is gone, no drain is needed — just clean up its leftover assignment state if any."],"exampleFix":"// before\nworker.drainIfNotInProgress(\"worker-3\"); // stale id\n// after\nif (!worker.getCurrentAvailableWorkers().contains(\"worker-3\")) {\n    log.warn(\"worker-3 is not active; nothing to drain\");\n    return;\n}\nworker.drainIfNotInProgress(\"worker-3\");","handlingStrategy":"validation","validationCode":"// confirm the worker is currently active\nif (!worker.getCurrentAvailableWorkers().contains(workerId)) {\n    throw new IllegalArgumentException(workerId + \" is not an active worker\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    worker.drainIfNotInProgress(workerId);\n} catch (UnknownWorkerException e) {\n    log.warn(\"Worker {} not active; refresh worker list\", workerId);\n}","preventionTips":["Fetch the live worker list and pick exact ids, not cached ones","Keep scripts in sync with configured workerId values","Skip drain if the worker already deregistered"],"tags":["pulsar-functions","drain","unknown-worker"],"backgroundTag":"unknown-worker-id","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"}