apache/pulsar · error · SchedulerManager.WorkerNotRemovedAfterPriorDrainException

Worker ${workerId} was not yet removed after a prior drain o

Error message

Worker ${workerId} was not yet removed after a prior drain op; try later

What it means

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.

Source

Thrown at pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/SchedulerManager.java:283

    public Future<?> drainIfNotInProgress(String workerId) {
        if (drainInProgressFlag.compareAndSet(false, true)) {
            try {
                Set<String> availableWorkers = getCurrentAvailableWorkers();
                if (availableWorkers.size() <= 1) {
                    throw new TooFewWorkersException();
                }

                // A worker must be specified at this point. This would be set up by the caller.
                Objects.requireNonNull(workerId);

                // [We can get stricter, and require that every drain op be followed up with a cleanup of the
                // corresponding worker before any other drain op, so that the drainOpStatusMap should be empty
                // at the next drain operation.]
                if (drainOpStatusMap.containsKey(workerId)) {
                    String warnString = "Worker " + workerId
                            + " was not removed yet from SchedulerManager after previous drain op";
                    log.warn(warnString);
                    throw new WorkerNotRemovedAfterPriorDrainException();
                }

                if (!availableWorkers.contains(workerId)) {
                    log.info().attr("workerId", workerId)
                            .log("invokeDrain was called for a worker which is not currently active");
                    throw new UnknownWorkerException();
                }

                return drain(workerId);
            } finally {
                drainInProgressFlag.set(false);
            }
        } else {
            throw new DrainInProgressException();
        }
    }

    public LongRunningProcessStatus getDrainStatus(String workerId) {

View on GitHub (pinned to 820761864e)

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.

Example fix

// before
worker.drainIfNotInProgress(workerId); // re-drain too early
// after
LongRunningProcessStatus st = worker.getDrainStatus(workerId);
if (st.getStatus() != LongRunningProcessStatus.Status.SUCCESS) {
    log.info("Prior drain not finished; retry later");
    return;
}
worker.drainIfNotInProgress(workerId);

Defensive patterns

Strategy: retry

Validate before calling

// check prior drain is finished and worker removed before re-draining
LongRunningProcessStatus prev = worker.getDrainStatus(workerId);
boolean alreadyDraining = prev != null
    && prev.getStatus() == LongRunningProcessStatus.Status.RUNNING;
if (alreadyDraining) {
    throw new IllegalStateException("Wait for prior drain of " + workerId + " to finish");
}

Try / catch

try {
    worker.drainIfNotInProgress(workerId);
} catch (WorkerNotRemovedAfterPriorDrainException e) {
    // poll getDrainStatus until previous drain completes, then retry
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of apache/pulsar@820761864e (2026-09-06). Data as JSON: /api/errors/947b61c601bc13f6. Report an issue: GitHub.