docs: @VisibleForTesting
    Pair<SortedMap<Interval, String>, Map<Interval, List<DataSegment>>> checkSegments()
    {
      // Pair<interval -> version, interval -> list<DataSegment>>
      Pair<Map<Interval, String>, Map<Interval, List<DataSegment>>> derivativeSegmentsSnapshot =
          getVersionAndBaseSegments(
              metadataStorageCoordinator.getUsedSegmentsForInterval(
                  dataSource,
                  ALL_INTERVAL
              )
          );
      // Pair<interval -> max(created_date), interval -> list<DataSegment>>
      Pair<Map<Interval, String>, Map<Interval, List<DataSegment>>> baseSegmentsSnapshot =
          getMaxCreateDateAndBaseSegments(
              metadataStorageCoordinator.getUsedSegmentAndCreatedDateForInterval(
                  spec.getBaseDataSource(),
                  ALL_INTERVAL
              )
          );
      // baseSegments are used to create HadoopIndexTask
      Map<Interval, List<DataSegment>> baseSegments = baseSegmentsSnapshot.rhs;
      Map<Interval, List<DataSegment>> derivativeSegments = derivativeSegmentsSnapshot.rhs;
      // use max created_date of base segments as the version of derivative segments
      Map<Interval, String> maxCreatedDate = baseSegmentsSnapshot.lhs;
      Map<Interval, String> derivativeVersion = derivativeSegmentsSnapshot.lhs;
      SortedMap<Interval, String> sortedToBuildInterval =
          new TreeMap<>(Comparators.intervalsByStartThenEnd().reversed());
      // find the intervals to drop and to build
      MapDifference<Interval, String> difference = Maps.difference(maxCreatedDate, derivativeVersion);
      Map<Interval, String> toBuildInterval = new HashMap<>(difference.entriesOnlyOnLeft());
      Map<Interval, String> toDropInterval = new HashMap<>(difference.entriesOnlyOnRight());
      // if some intervals are in running tasks and the versions are the same, remove it from toBuildInterval
      // if some intervals are in running tasks, but the versions are different, stop the task. 
      for (Interval interval : runningVersion.keySet()) {
        if (toBuildInterval.containsKey(interval) 
            && toBuildInterval.get(interval).equals(runningVersion.get(interval))
            ) {
          toBuildInterval.remove(interval);
  
        } else if (
            toBuildInterval.containsKey(interval) 
            && !toBuildInterval.get(interval).equals(runningVersion.get(interval))
        ) {
          if (taskMaster.getTaskQueue().isPresent()) {
            taskMaster.getTaskQueue().get().shutdown(runningTasks.get(interval).getId(), "version mismatch");
            runningTasks.remove(interval);
          }
        }
      }
      // drop derivative segments which interval equals the interval in toDeleteBaseSegments 
      for (Interval interval : toDropInterval.keySet()) {
        for (DataSegment segment : derivativeSegments.get(interval)) {
          segmentManager.removeSegment(segment.getId());
        }
      }
      // data of the latest interval will be built firstly.
      sortedToBuildInterval.putAll(toBuildInterval);
      return new Pair<>(sortedToBuildInterval, baseSegments);
    }