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); 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); 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; } DistcpBaseService(ConduitConfig config, String name,
      Cluster srcCluster, Cluster destCluster, Cluster currentCluster,
      CheckpointProvider provider, Set<String> streamsToProcess); @Override long getMSecondsTillNextRun(long currentTime); Cluster getCurrentCluster(); static void createListing(FileSystem fs, FileStatus fileStatus,
      List<FileStatus> results);  }
@Test(priority = 2) public void testNegative() throws Exception { cleanUP(); createInvalidData(); Map<String, FileStatus> fileCopyListMap = mirrorService .getDistCPInputFile(); assert fileCopyListMap.size() == 0; }