class DistcpBaseService extends AbstractService {

  protected Map<String, FileStatus> getDistCPInputFile()
      throws Exception {
    Map<String,FileStatus> result = new HashMap<String, FileStatus>();
    for (String stream : streamsToProcess) {
      int pathsAlreadyAdded = 0;
      LOG.info("Processing stream " + stream);
      byte[] value = provider.read(getCheckPointKey(stream));
      Path inputPath = new Path(getInputPath(), stream);
      Path lastCheckPointPath = null;
      Path nextPath = null;
      List<FileStatus> filesLastCopiedDir;
      if (value != null) {
        String checkPointValue = new String(value);
        // creating a path object from empty string throws exception;hence
        // checking for it
        if (!checkPointValue.trim().equals("")) {
          lastCheckPointPath = new Path(checkPointValue);
        }
        lastCheckPointPath = fullyQualifyCheckPointWithReadURL
            (lastCheckPointPath, srcCluster);
        if (lastCheckPointPath == null
            || !getSrcFs().exists(lastCheckPointPath)) {
          LOG.warn("Invalid checkpoint found [" + lastCheckPointPath
              + "] for stream " + stream + ";Ignoring it");
        } else {
          Date lastDate = CalendarHelper.getDateFromStreamDir(inputPath,
              lastCheckPointPath);
          nextPath = CalendarHelper.getNextMinutePathFromDate(lastDate,
              inputPath);
        }

      }
      if (nextPath == null) {
        filesLastCopiedDir = new ArrayList<FileStatus>();
        LOG.info("Finding the starting directoryfor stream [" + stream + "]");
        nextPath = getStartingDirectory(stream, filesLastCopiedDir);
        if (nextPath == null) {
          LOG.debug("No start directory found,returning the empty result");
          continue;
        }
        LOG.debug("Uncopied Files from directory last copied are "
            + FileUtil.toStringOfFileStatus(filesLastCopiedDir));
        for (FileStatus fileStatus : filesLastCopiedDir) {
          String destnPath = getFinalDestinationPath(fileStatus);
          if (destnPath != null) {
            LOG.info("Adding to input of Distcp.Move [" + fileStatus.getPath()
                + "] to " + destnPath);
            result.put(destnPath, fileStatus);
          }
        }
      }
      LOG.info("Starting directory for stream [" + stream + "]" + " is ["
          + nextPath + "]");
      Date nextDate = CalendarHelper.getDateFromStreamDir(inputPath, nextPath);
      // if next to next path exist than only add the next path so that the path
      // being added to disctp input is not the current path
      Path nextToNextPath = CalendarHelper.getNextMinutePathFromDate(nextDate,
          inputPath);
      Path lastPathAdded = null;
      FileStatus[] nextPathFileStatus;
      while (pathsAlreadyAdded <= numOfDirPerDistcpPerStream
          && srcFs.exists(nextToNextPath)
          && (nextPathFileStatus = FileUtil
          .listStatusAsPerHDFS(srcFs, nextPath)) != null) {
        if(nextPathFileStatus.length==0){
          LOG.info(nextPath + " is an empty directory");
          FileStatus srcFileStatus = srcFs.getFileStatus(nextPath); 
          String destnPath = getFinalDestinationPath(srcFileStatus);
          if(destnPath!=null){
            LOG.info("Adding to input of Distcp.Move ["+nextPath+"] to "+destnPath);
            result.put(destnPath,srcFileStatus);
          }
        }
        else{
          for(FileStatus fStatus:nextPathFileStatus){
            String destnPath = getFinalDestinationPath(fStatus);
            if(destnPath!=null){
              LOG.info("Adding to input of Distcp.Move ["+fStatus.getPath()+"] to "+destnPath);
              result.put(destnPath,fStatus);
            }
          }
        } 
        pathsAlreadyAdded++;
        lastPathAdded = nextPath;
        nextPath = nextToNextPath;
        nextDate = CalendarHelper.addAMinute(nextDate);
        nextToNextPath = CalendarHelper.getNextMinutePathFromDate(nextDate,
            inputPath);
      }
      if (lastPathAdded != null) {
        checkPointPaths.put(stream, lastPathAdded);
        Date lastDateAdded = CalendarHelper.getDateFromStreamDir(inputPath,
            lastPathAdded);
        lastProcessedFile.put(stream, lastDateAdded.getTime());
      }

    }
    return result;
  }

  public  DistcpBaseService(ConduitConfig config, String name,
      Cluster srcCluster, Cluster destCluster, Cluster currentCluster,
      CheckpointProvider provider, Set<String> streamsToProcess);

}