code: public Set<ResourceWorkerSlot> getKeepAssign(DefaultTopologyAssignContext defaultContext, Set<Integer> needAssigns) {
          Set<Integer> keepAssignIds = new HashSet<>();
          keepAssignIds.addAll(defaultContext.getAllTaskIds());
          keepAssignIds.removeAll(defaultContext.getUnstoppedTaskIds());
          keepAssignIds.removeAll(needAssigns);
          Set<ResourceWorkerSlot> keeps = new HashSet<>();
          if (keepAssignIds.isEmpty()) {
              return keeps;
          }
  
          Assignment oldAssignment = defaultContext.getOldAssignment();
          if (oldAssignment == null) {
              return keeps;
          }
          keeps.addAll(defaultContext.getOldWorkers());
          for (ResourceWorkerSlot worker : defaultContext.getOldWorkers()) {
              for (Integer task : worker.getTasks()) {
                  if (!keepAssignIds.contains(task)) {
                      keeps.remove(worker);
                      break;
                  }
              }
          }
          return keeps;
      }
docs: Get the task Map which the task is alive and will be kept only when type is ASSIGN_TYPE_MONITOR, it is valid
  
  @param defaultContext default topology context
  @param needAssigns    a set of tasks to be assigned
  @return a set of assigned slots