{"record":{"id":"10ab4296412e1590","repo":"nathanmarz/storm","slug":"slot-slot-getnodeid-slot-getport-is-already-occupied","errorCode":null,"errorMessage":"slot: [${slot.getNodeId()}, ${slot.getPort()}] is already occupied.","messagePattern":"slot: \\[(.+?), (.+?)\\] is already occupied\\.","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"storm-core/src/jvm/backtype/storm/scheduler/Cluster.java","lineNumber":283,"sourceCode":"        SchedulerAssignment assignment = this.getAssignmentById(topology.getId());\n        if (topology == null || assignment == null) {\n            return 0;\n        }\n\n        Set<WorkerSlot> slots = new HashSet<WorkerSlot>();\n        slots.addAll(assignment.getExecutorToSlot().values());\n\n        return slots.size();\n    }\n\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    /**","sourceCodeStart":265,"sourceCodeEnd":301,"githubUrl":"https://github.com/nathanmarz/storm/blob/cdb116e942666973bc4eaa0df098d5bab82739e7/storm-core/src/jvm/backtype/storm/scheduler/Cluster.java#L265-L301","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Before assigning, check cluster.getUsedSlots()/isSlotOccupied(slot) and choose a free slot instead.","If the existing assignment is stale (topology dead or being rebalanced), call cluster.freeSlot(slot) (or unassign the old topology) before assign.","Compute candidate slots from slotsAvailableOnHosts / getUnusedSlots rather than hardcoding node:port values.","If topologyId matches the topology already occupying the slot, use assign's per-assignment update path (getAssignmentById) instead of treating it as free."],"exampleFix":"// before\ncluster.assign(new WorkerSlot(\"node1\", 6700), topologyId, executors); // may already be occupied\n// after\nif (!cluster.isSlotOccupied(new WorkerSlot(\"node1\", 6700))) {\n    cluster.assign(new WorkerSlot(\"node1\", 6700), topologyId, executors);\n} else {\n    cluster.freeSlot(new WorkerSlot(\"node1\", 6700));\n    cluster.assign(new WorkerSlot(\"node1\", 6700), topologyId, executors);\n}","handlingStrategy":"validation","validationCode":"WorkerSlot slot = ...;\nif (cluster.isSlotOccupied(slot)) {\n    // pick another slot or free the stale one first\n    slot = cluster.getUsedSlots().isEmpty() ? null : null; // choose from cluster.getAvailableSlots / slotsAvailableOnHosts\n}","typeGuard":null,"tryCatchPattern":"try {\n    cluster.assign(slot, topologyId, executors);\n} catch (RuntimeException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"is already occupied\")) {\n        cluster.freeSlot(slot);\n        cluster.assign(slot, topologyId, executors);\n    } else throw e;\n}","preventionTips":["Always derive candidate slots from cluster.getUnusedSlots()/getAvailableSlots(hosts), never hardcode node:port.","Call freeSlot (or unassign) before reassigning slots of dead/rebalanced topologies.","In custom schedulers, check isSlotOccupied immediately before every assign call.","Clear stale assignments after supervisor/node restarts before scheduling."],"tags":["scheduler","worker-slot","conflict","storm"],"backgroundTag":"address-already-in-use","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"}