nathanmarz/storm · error · RuntimeException

the executor is already assigned, you should unassign it…

Error message

the executor is already assigned, you should unassign it before assign it to another slot.

What it means

Apache Storm's Cluster.assign() refuses to assign executors to a worker slot when one or more of those executors already have an assignment for the given topology. The library enforces the invariant that an executor can occupy only one slot at a time; callers must first unassign (or release) the existing assignment before re-assigning. It is a plain RuntimeException signaling a scheduler-state misuse.

Solutions

1. Before calling assign(), unassign the executors' current assignment: cluster.unassignById(assignmentId) or assignment.getUnassignedExecutors(), or call cluster.assign only with executors from assignment.getUnassignedExecutors().
2. Filter the executor set: executors.removeAll(assignment.getExecutors()) so only not-yet-assigned executors are passed to assign().
3. If the intent is to move an executor to another slot, mark the assignment dirty / free the executor (e.g., via cluster.freeSlot(slot) or unassign) then re-assign.
4. Check your scheduler code for duplicate assignment of the same ExecutorDetails in one scheduling cycle.

Example fix

// before
cluster.assign(topologyId, newSlot, allExecutors);
// after
Set<ExecutorDetails> unassigned = assignment != null ? assignment.getUnassignedExecutors() : allExecutors;
if (!unassigned.isEmpty()) {
    cluster.assign(topologyId, newSlot, unassigned);
}

// Java
Set<ExecutorDetails> alreadyAssigned = assignment != null ? assignment.getExecutors() : Collections.emptySet();
boolean conflict = executors.stream().anyMatch(alreadyAssigned::contains);
if (!conflict) cluster.assign(topologyId, slot, executors);

boolean isReassignSafe(SchedulerAssignment a, Collection<ExecutorDetails> exec) {
    return a == null || exec.stream().noneMatch(a::isExecutorAssigned);
}

When it happens

Trigger: Calling Cluster.assign(topologyId, slot, executors) where assignment != null for topologyId and any executor in `executors` returns true from SchedulerAssignment.isExecutorAssigned(). Typically: reassigning an executor without calling unassignById/unassign first, or assigning overlapping executor sets in two consecutive assign() calls.

Common situations: Custom IScheduler implementations that reassign executors on rebalance/scale-out without clearing prior assignments; calling assign() twice for the same executor during topology rescheduling; bugs in scheduler plugins that assume assign() overwrites existing assignments instead of throwing.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• slot: [ , ] is already occupied.
• Each element of the list
• Field must be an Iterable of
• Field must be a power of 2.
• Topology with name ` ` already exists on cluster

AI-assisted analysis of nathanmarz/storm@cdb116e942 (2026-09-12). Data as JSON: /api/errors/7f5f29aa7d3491b3. Report an issue: GitHub.

Appendix: source

Thrown at storm-core/src/jvm/backtype/storm/scheduler/Cluster.java:293

    /**
     * Assign the slot to the executors for this topology.
     * 
     * @throws RuntimeException if the specified slot is already occupied.
     */
    public void assign(WorkerSlot slot, String topologyId, Collection<ExecutorDetails> executors) {
        if (this.isSlotOccupied(slot)) {
            throw new RuntimeException("slot: [" + slot.getNodeId() + ", " + slot.getPort() + "] is already occupied.");
        }
        
        SchedulerAssignmentImpl assignment = (SchedulerAssignmentImpl)this.getAssignmentById(topologyId);
        if (assignment == null) {
            assignment = new SchedulerAssignmentImpl(topologyId, new HashMap<ExecutorDetails, WorkerSlot>());
            this.assignments.put(topologyId, assignment);
        } else {
            for (ExecutorDetails executor : executors) {
                 if (assignment.isExecutorAssigned(executor)) {
                     throw new RuntimeException("the executor is already assigned, you should unassign it before assign it to another slot.");
                 }
            }
        }

        assignment.assign(slot, executors);
    }

    /**
     * Gets all the available slots in the cluster.
     * 
     * @return
     */
    public List<WorkerSlot> getAvailableSlots() {
        List<WorkerSlot> slots = new ArrayList<WorkerSlot>();
        for (SupervisorDetails supervisor : this.supervisors.values()) {
            slots.addAll(this.getAvailableSlots(supervisor));
        }

View on GitHub (pinned to cdb116e942)