code: 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 un-assign it first " +
                              "before assign it to another slot.");
                  }
              }
          }
          assignment.assign(slot, executors);
      }
docs: Assign the slot to the executors for this topology.
  
  @throws RuntimeException if the specified slot is already occupied.