docs: @VisibleForTesting
    void checkSegmentsAndSubmitTasks()
    {
      synchronized (taskLock) {
        List<Interval> intervalsToRemove = new ArrayList<>();
        for (Map.Entry<Interval, HadoopIndexTask> entry : runningTasks.entrySet()) {
          Optional<TaskStatus> taskStatus = taskStorage.getStatus(entry.getValue().getId());
          if (!taskStatus.isPresent() || !taskStatus.get().isRunnable()) {
            intervalsToRemove.add(entry.getKey());
          }
        }
        for (Interval interval : intervalsToRemove) {
          runningTasks.remove(interval);
          runningVersion.remove(interval);
        }
  
        if (runningTasks.size() == maxTaskCount) {
          //if the number of running tasks reach the max task count, supervisor won't submit new tasks.
          return;
        }
        Pair<SortedMap<Interval, String>, Map<Interval, List<DataSegment>>> toBuildIntervalAndBaseSegments = checkSegments();
        SortedMap<Interval, String> sortedToBuildVersion = toBuildIntervalAndBaseSegments.lhs;
        Map<Interval, List<DataSegment>> baseSegments = toBuildIntervalAndBaseSegments.rhs;
        missInterval = sortedToBuildVersion.keySet();
        submitTasks(sortedToBuildVersion, baseSegments);
      }
    }