LocalStreamService extends AbstractService implements
ConfigConstants { Map<Path, Path> populateTrashCommitPaths(Set<FileStatus> trashSet) { Map<Path, Path> trashPaths = new TreeMap<Path, Path>(); Path trash = srcCluster.getTrashPathWithDateHour(); Iterator<FileStatus> it = trashSet.iterator(); while (it.hasNext()) { FileStatus src = it.next(); Path target = null; target = new Path(trash, src.getPath().getParent().getName() + "-" + src.getPath().getName()); LOG.debug("Trashing [" + src.getPath() + "] to [" + target + "]"); trashPaths.put(src.getPath(), target); } return trashPaths; } LocalStreamService(ConduitConfig config, Cluster srcCluster,
      Cluster currentCluster, CheckpointProvider provider,
      Set<String> streamsToProcess); @Override long getMSecondsTillNextRun(long currentTime); void createListing(FileSystem fs, FileStatus fileStatus,
      Map<FileStatus, String> results, Set<FileStatus> trashSet,
      Table<String, String, String> checkpointPaths); long getLastAddedPartTime(String stream); Cluster getCurrentCluster(); String getTopicNameFromDestnPath(Path destnPath); @Override String getServiceType();  }
@Test public void testPopulateTrashPaths() throws Exception { FileStatus[] status = new FileStatus[10]; String[] expectedstatus = new String[10]; status[0] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test1/testcluster1/test1-2012-08-29-07-09_00000")); status[1] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test1/testcluster1/test1-2012-08-29-07-04_00000")); status[2] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test2/testcluster1/test2-2012-08-29-07-09_00003")); status[3] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test1/testcluster2/test1-2012-08-13-07-09_00000")); status[4] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test1/testcluster1/test1-2012-08-29-07-09_00009")); status[5] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test1/testcluster1/test1-2012-08-29-07-12_00000")); status[6] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test1/testcluster1/test1-2012-08-29-07-10_00000")); status[7] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test2/testcluster1/test2-2012-08-29-07-45_00000")); status[8] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test1/testcluster2/test1-2012-08-29-07-09_00078")); status[9] = new FileStatus(20, false, 3, 23823, 2438232, new Path( "/conduit/data/test1/testcluster2/test1-2012-08-29-07-04_00034")); expectedstatus[0] = "/conduit/data/test1/testcluster1/test1-2012-08-29-07-04_00000"; expectedstatus[1] = "/conduit/data/test1/testcluster1/test1-2012-08-29-07-09_00000"; expectedstatus[2] = "/conduit/data/test1/testcluster1/test1-2012-08-29-07-09_00009"; expectedstatus[3] = "/conduit/data/test1/testcluster1/test1-2012-08-29-07-10_00000"; expectedstatus[4] = "/conduit/data/test1/testcluster1/test1-2012-08-29-07-12_00000"; expectedstatus[5] = "/conduit/data/test1/testcluster2/test1-2012-08-13-07-09_00000"; expectedstatus[6] = "/conduit/data/test1/testcluster2/test1-2012-08-29-07-04_00034"; expectedstatus[7] = "/conduit/data/test1/testcluster2/test1-2012-08-29-07-09_00078"; expectedstatus[8] = "/conduit/data/test2/testcluster1/test2-2012-08-29-07-09_00003"; expectedstatus[9] = "/conduit/data/test2/testcluster1/test2-2012-08-29-07-45_00000"; Set<FileStatus> trashSet = new HashSet<FileStatus>(); for (int i = 0; i < 10; ++i) { trashSet.add(status[i]); } ConduitConfig conduitConfig = buildTestConduitConfig(); Cluster cluster = ClusterTest.buildLocalCluster(); Set<String> streamsToProcess = new HashSet<String>(); streamsToProcess.addAll(conduitConfig.getSourceStreams().keySet()); TestLocalStreamService service = new TestLocalStreamService( conduitConfig, cluster, null, new FSCheckpointProvider( cluster.getCheckpointDir()), streamsToProcess); Map<Path, Path> trashCommitPaths = service .populateTrashCommitPaths(trashSet); Set<Path> srcPaths = trashCommitPaths.keySet(); Iterator<Path> it = srcPaths.iterator(); int i = 0; while (it.hasNext()) { String actualPath = it.next().toString(); String expectedPath = expectedstatus[i]; LOG.debug("Comparing Trash Paths Actual [" + actualPath + "] Expected [" + expectedPath + "]"); Assert.assertEquals(actualPath, expectedPath); i++; } Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.RETRY_MKDIR,"stream1").getValue().longValue() , 0); Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.RETRY_CHECKPOINT,"stream1").getValue().longValue() , 0); Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.RETRY_RENAME,"stream1").getValue().longValue() , 0); Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.FILES_COPIED_COUNT,"stream1").getValue().longValue() , 0); }