nathanmarz/storm · error · RuntimeException

slot: [ , ] is already occupied.

Error message

slot: [${slot.getNodeId()}, ${slot.getPort()}] is already occupied.

What it means

IScheduler implementations assign executors to worker slots via Cluster.assign. Before writing the assignment, it checks isSlotOccupied; if the node:port slot already belongs to another topology's assignment, it throws, because two topologies must never share one worker slot. The Javadoc explicitly documents this RuntimeException for occupied slots.

Solutions

1. Before assigning, check cluster.getUsedSlots()/isSlotOccupied(slot) and choose a free slot instead.
2. If the existing assignment is stale (topology dead or being rebalanced), call cluster.freeSlot(slot) (or unassign the old topology) before assign.
3. Compute candidate slots from slotsAvailableOnHosts / getUnusedSlots rather than hardcoding node:port values.
4. If topologyId matches the topology already occupying the slot, use assign's per-assignment update path (getAssignmentById) instead of treating it as free.

Example fix

// before
cluster.assign(new WorkerSlot("node1", 6700), topologyId, executors); // may already be occupied
// after
if (!cluster.isSlotOccupied(new WorkerSlot("node1", 6700))) {
    cluster.assign(new WorkerSlot("node1", 6700), topologyId, executors);
} else {
    cluster.freeSlot(new WorkerSlot("node1", 6700));
    cluster.assign(new WorkerSlot("node1", 6700), topologyId, executors);
}

WorkerSlot slot = ...;
if (cluster.isSlotOccupied(slot)) {
    // pick another slot or free the stale one first
    slot = cluster.getUsedSlots().isEmpty() ? null : null; // choose from cluster.getAvailableSlots / slotsAvailableOnHosts
}

try {
    cluster.assign(slot, topologyId, executors);
} catch (RuntimeException e) {
    if (e.getMessage() != null && e.getMessage().contains("is already occupied")) {
        cluster.freeSlot(slot);
        cluster.assign(slot, topologyId, executors);
    } else throw e;
}

When it happens

Trigger: A custom IScheduler calls cluster.assign(slot, topologyId, executors) with a WorkerSlot that isSlotOccupied returns true for — the slot is already used by a different topology's existing SchedulerAssignment (typically slot for a topology still marked as assigned but whose worker died or wasn't cleaned up).

Common situations: Custom scheduler logic that doesn't call freeSlot/getUsedSlots before assigning; supervisor/node restart left stale assignments so the scheduler reuses an occupied port; scheduling topology B onto slots computed for topology A after topology A's assignment wasn't released.

Related errors

• the executor is already assigned, you should unassign it…
• 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/10ab4296412e1590. Report an issue: GitHub.

Appendix: source

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

        SchedulerAssignment assignment = this.getAssignmentById(topology.getId());
        if (topology == null || assignment == null) {
            return 0;
        }

        Set<WorkerSlot> slots = new HashSet<WorkerSlot>();
        slots.addAll(assignment.getExecutorToSlot().values());

        return slots.size();
    }

    /**
     * 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);
    }

    /**

View on GitHub (pinned to cdb116e942)