apache/pulsar · warning · SchedulerManager.DrainInProgressException

Another drain is in progress

Error message

Another drain is in progress

What it means

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.

Source

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

                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) {
        long startTime = System.nanoTime();
        LongRunningProcessStatus status = Optional.ofNullable(workerId).map(id ->
                Optional.ofNullable(drainOpStatusMap.get(id)).map(opStatus ->
                        switch (opStatus) {
                            case DrainCompleted ->
                                    LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.SUCCESS);
                            case DrainInProgress ->
                                    LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.RUNNING);
                            case DrainNotInProgress ->
                                    LongRunningProcessStatus.forStatus(LongRunningProcessStatus.Status.NOT_RUN);
                        }).orElse(
                        LongRunningProcessStatus.forError("Worker " + id + " not found in drain records")
                )
        ).orElse(

View on GitHub (pinned to 820761864e)

Solutions

1. Run drain operations sequentially: wait for each drain to complete (poll getDrainStatus) before starting the next.
2. Catch DrainInProgressException and retry with backoff.
3. Refactor automation to a single-threaded drain queue.

Example fix

// before
workers.forEach(w -> worker.drainIfNotInProgress(w)); // parallel drains
// after
for (String w : workers) {
    worker.drainIfNotInProgress(w);
    while (worker.getDrainStatus(w).getStatus() == LongRunningProcessStatus.Status.RUNNING) {
        Thread.sleep(1000);
    }
}

try {
    worker.drainIfNotInProgress(workerId);
} catch (DrainInProgressException e) {
    // another drain is running: wait, then retry with backoff
}

When it happens

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

Common situations: Automated rolling shutdown that drains multiple workers in parallel instead of sequentially; slow drain (large assignment transfer) overlapping with the next drain request.

Related errors

• Rebalance already in progress
• Worker ${workerId} was not yet removed after a prior drain o
• Worker ${workerId} is not among the current workers in the s
• Another drain is in progress
• Concurrent modification

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