apache/pulsar · warning · RestException

Another drain is in progress

Error message

Another drain is in progress

What it means

This HTTP 409 error is returned by the drain endpoint when a drain operation is already running. SchedulerManager.drainIfNotInProgress() enforces a single concurrent drain cluster-wide to avoid conflicting function-instance migrations while a worker is being decommissioned.

Solutions

1. Wait for the current drain to complete before starting another
2. Poll the drain/status endpoint or watch worker logs for drain completion
3. Serialize drain requests in automation (no concurrent drain calls)

Example fix

// before: fire-and-forget loop
for (String w : workers) { admin.functions().drain(w); }
// after: sequential with conflict handling
for (String w : workers) {
    try {
        admin.functions().drain(w);
    } catch (PulsarAdminException e) {
        if (e.getStatusCode() == 409) { /* wait for in-progress drain, retry */ }
    }
}

// no pre-check API for in-progress drain; serialize drains in your tooling

try {
    admin.functions().drain(workerId);
} catch (PulsarAdminException e) {
    if (e.getStatusCode() == 409 && e.getMessage().contains("Another drain is in progress")) {
        // wait for current drain to finish, then retry
    }
}

When it happens

Trigger: Calling PUT /admin/v2/worker/drain/{workerId} (or pulsar-admin functions worker drain) while a prior drain on the leader is still in progress.

Common situations: Draining multiple workers back-to-back without waiting for each drain to finish; automation that retries drain on timeout; large function counts making drains slow.

Related errors

• Another drain is in progress
• Concurrent modification
• Rebalance already in progress
• Worker is not among the current workers in the system
• Worker was not yet removed after a prior drain op; try later

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

Appendix: source

Thrown at pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/WorkerImpl.java:274

                .attr("workerId", workerId)
                .attr("clientRole", authParams.getClientRole())
                .attr("originalPrincipal", authParams.getOriginalPrincipal())
                .attr("calledOnLeaderUri", calledOnLeaderUri)
                .attr("actualWorkerId", actualWorkerId)
                .log("drain called");

        throwIfNotSuperUser(authParams, "drain worker");

        // Depending on which operations we decide to allow, we may add checks here to error/exception if
        //      calledOnLeaderUri is true on a non-leader
        //      calledOnLeaderUri is false on a leader
        // For now, deal with everything.

        if (worker().getLeaderService().isLeader()) {
            try {
                worker().getSchedulerManager().drainIfNotInProgress(workerId);
            } catch (SchedulerManager.DrainInProgressException e) {
                throw new RestException(Status.CONFLICT, "Another drain is in progress");
            } catch (SchedulerManager.TooFewWorkersException e) {
                throw new RestException(Status.BAD_REQUEST, "Too few workers (need at least 2)");
            } catch (SchedulerManager.WorkerNotRemovedAfterPriorDrainException e) {
                String errString = "Worker " + workerId + " was not yet removed after a prior drain op; try later";
                throw new RestException(Status.PRECONDITION_FAILED, errString);
            } catch (SchedulerManager.UnknownWorkerException e) {
                String errString = "Worker " + workerId + " is not among the current workers in the system";
                throw new RestException(Status.BAD_REQUEST, errString);
            }
        } else {
            URI redirect = buildRedirectUriForDrainRelatedOp(uri, workerId);
            log.info().attr("redirect", redirect).log("Not leader; redirect URI=");
            throw new WebApplicationException(Response.temporaryRedirect(redirect).build());
        }
    }

    @Override
    public LongRunningProcessStatus getDrainStatus(final URI uri, final String inWorkerId,

View on GitHub (pinned to 820761864e)