{"record":{"id":"7f5f29aa7d3491b3","repo":"nathanmarz/storm","slug":"the-executor-is-already-assigned-you-should-unassign-it","errorCode":null,"errorMessage":"the executor is already assigned, you should unassign it before assign it to another slot.","messagePattern":"the executor is already assigned, you should unassign it before assign it to another slot\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/backtype/storm/scheduler/Cluster.java","lineNumber":293,"sourceCode":"\n    /**\n     * Assign the slot to the executors for this topology.\n     * \n     * @throws RuntimeException if the specified slot is already occupied.\n     */\n    public void assign(WorkerSlot slot, String topologyId, Collection<ExecutorDetails> executors) {\n        if (this.isSlotOccupied(slot)) {\n            throw new RuntimeException(\"slot: [\" + slot.getNodeId() + \", \" + slot.getPort() + \"] is already occupied.\");\n        }\n        \n        SchedulerAssignmentImpl assignment = (SchedulerAssignmentImpl)this.getAssignmentById(topologyId);\n        if (assignment == null) {\n            assignment = new SchedulerAssignmentImpl(topologyId, new HashMap<ExecutorDetails, WorkerSlot>());\n            this.assignments.put(topologyId, assignment);\n        } else {\n            for (ExecutorDetails executor : executors) {\n                 if (assignment.isExecutorAssigned(executor)) {\n                     throw new RuntimeException(\"the executor is already assigned, you should unassign it before assign it to another slot.\");\n                 }\n            }\n        }\n\n        assignment.assign(slot, executors);\n    }\n\n    /**\n     * Gets all the available slots in the cluster.\n     * \n     * @return\n     */\n    public List<WorkerSlot> getAvailableSlots() {\n        List<WorkerSlot> slots = new ArrayList<WorkerSlot>();\n        for (SupervisorDetails supervisor : this.supervisors.values()) {\n            slots.addAll(this.getAvailableSlots(supervisor));\n        }\n","sourceCodeStart":275,"sourceCodeEnd":311,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java#L275-L311","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Before calling assign(), unassign the executors' current assignment: cluster.unassignById(assignmentId) or assignment.getUnassignedExecutors(), or call cluster.assign only with executors from assignment.getUnassignedExecutors().","Filter the executor set: executors.removeAll(assignment.getExecutors()) so only not-yet-assigned executors are passed to assign().","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.","Check your scheduler code for duplicate assignment of the same ExecutorDetails in one scheduling cycle."],"exampleFix":"// before\ncluster.assign(topologyId, newSlot, allExecutors);\n// after\nSet<ExecutorDetails> unassigned = assignment != null ? assignment.getUnassignedExecutors() : allExecutors;\nif (!unassigned.isEmpty()) {\n    cluster.assign(topologyId, newSlot, unassigned);\n}","handlingStrategy":"validation","validationCode":"// Java\nSet<ExecutorDetails> alreadyAssigned = assignment != null ? assignment.getExecutors() : Collections.emptySet();\nboolean conflict = executors.stream().anyMatch(alreadyAssigned::contains);\nif (!conflict) cluster.assign(topologyId, slot, executors);","typeGuard":"boolean isReassignSafe(SchedulerAssignment a, Collection<ExecutorDetails> exec) {\n    return a == null || exec.stream().noneMatch(a::isExecutorAssigned);\n}","tryCatchPattern":null,"preventionTips":["Always assign only executors from assignment.getUnassignedExecutors().","In custom schedulers, unassign/free existing assignments before re-assigning executors.","Add an assertion/test that each executor appears in exactly one WorkerSlot per scheduling round."],"tags":["scheduler","invalid-state-transition","storm"],"backgroundTag":"invalid-state-transition","analyzedSha":"cdb116e942666973bc4eaa0df098d5bab82739e7","analyzedAt":"2026-09-12T14:30:00.714Z","contentChangedAt":"2026-09-12T14:30:00.714Z","schemaVersion":2},"datasetVersion":"2026-09-16T09:17:16.951Z"}