test_class.declarator: LOG = Logger.getLogger(LocalStreamServiceTest.class)
test_class.modifier: private static
test_class.original_string: private static Logger LOG = Logger.getLogger(LocalStreamServiceTest.class);
test_class.type: Logger
test_class.var_name: LOG
test_class.declarator: number_files = 9
test_class.modifier: private final static
test_class.original_string: private final static int number_files = 9;
test_class.type: int
test_class.var_name: number_files
test_class.declarator: expectedResults = new LinkedHashSet<String>()
test_class.original_string: Set<String> expectedResults = new LinkedHashSet<String>();
test_class.type: Set<String>
test_class.var_name: expectedResults
test_class.declarator: expectedTrashPaths = new LinkedHashSet<String>()
test_class.original_string: Set<String> expectedTrashPaths = new LinkedHashSet<String>();
test_class.type: Set<String>
test_class.var_name: expectedTrashPaths
test_class.declarator: expectedCheckPointPaths = new HashMap<String, String>()
test_class.original_string: Map<String, String> expectedCheckPointPaths = new HashMap<String, String>();
test_class.type: Map<String, String>
test_class.var_name: expectedCheckPointPaths
test_class.file: databus-worker/src/test/java/com/inmobi/databus/local/LocalStreamServiceTest.java
test_class.identifier: LocalStreamServiceTest
test_class.superclass: extends TestMiniClusterUtil
test_case.body: @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(
          "/databus/data/test1/testcluster1/test1-2012-08-29-07-09_00000"));
      status[1] = new FileStatus(20, false, 3, 23823, 2438232, new Path(
          "/databus/data/test1/testcluster1/test1-2012-08-29-07-04_00000"));
      status[2] = new FileStatus(20, false, 3, 23823, 2438232, new Path(
          "/databus/data/test2/testcluster1/test2-2012-08-29-07-09_00003"));
      status[3] = new FileStatus(20, false, 3, 23823, 2438232, new Path(
          "/databus/data/test1/testcluster2/test1-2012-08-13-07-09_00000"));
      status[4] = new FileStatus(20, false, 3, 23823, 2438232, new Path(
          "/databus/data/test1/testcluster1/test1-2012-08-29-07-09_00009"));
      status[5] = new FileStatus(20, false, 3, 23823, 2438232, new Path(
          "/databus/data/test1/testcluster1/test1-2012-08-29-07-12_00000"));
      status[6] = new FileStatus(20, false, 3, 23823, 2438232, new Path(
          "/databus/data/test1/testcluster1/test1-2012-08-29-07-10_00000"));
      status[7] = new FileStatus(20, false, 3, 23823, 2438232, new Path(
          "/databus/data/test2/testcluster1/test2-2012-08-29-07-45_00000"));
      status[8] = new FileStatus(20, false, 3, 23823, 2438232, new Path(
          "/databus/data/test1/testcluster2/test1-2012-08-29-07-09_00078"));
      status[9] = new FileStatus(20, false, 3, 23823, 2438232, new Path(
          "/databus/data/test1/testcluster2/test1-2012-08-29-07-04_00034"));
      
      expectedstatus[0] = "/databus/data/test1/testcluster1/test1-2012-08-29-07-04_00000";
      expectedstatus[1] = "/databus/data/test1/testcluster1/test1-2012-08-29-07-09_00000";
      expectedstatus[2] = "/databus/data/test1/testcluster1/test1-2012-08-29-07-09_00009";
      expectedstatus[3] = "/databus/data/test1/testcluster1/test1-2012-08-29-07-10_00000";
      expectedstatus[4] = "/databus/data/test1/testcluster1/test1-2012-08-29-07-12_00000";
      
      expectedstatus[5] = "/databus/data/test1/testcluster2/test1-2012-08-13-07-09_00000";
      expectedstatus[6] = "/databus/data/test1/testcluster2/test1-2012-08-29-07-04_00034";
      expectedstatus[7] = "/databus/data/test1/testcluster2/test1-2012-08-29-07-09_00078";
      
      
      expectedstatus[8] = "/databus/data/test2/testcluster1/test2-2012-08-29-07-09_00003";
      expectedstatus[9] = "/databus/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]);
      }
      
      Cluster cluster = ClusterTest.buildLocalCluster();
      TestLocalStreamService service = new TestLocalStreamService(
          buildTestDatabusConfig(), cluster, new FSCheckpointProvider(
              cluster.getCheckpointDir()));
      
      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++;
      }
    }
test_case.class_method_signature: LocalStreamServiceTest.testPopulateTrashPaths()
test_case.full_signature: @Test public void testPopulateTrashPaths()
test_case.identifier: testPopulateTrashPaths
test_case.invocations: add buildLocalCluster buildTestDatabusConfig getCheckpointDir populateTrashCommitPaths keySet iterator hasNext toString next debug assertEquals
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testPopulateTrashPaths()
focal_class.declarator: LOG = LogFactory.getLog(LocalStreamService.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Log LOG = LogFactory.getLog(LocalStreamService.class);
focal_class.type: Log
focal_class.var_name: LOG
focal_class.declarator: cluster
focal_class.modifier: private final
focal_class.original_string: private final Cluster cluster;
focal_class.type: Cluster
focal_class.var_name: cluster
focal_class.declarator: tmpPath
focal_class.modifier: private
focal_class.original_string: private Path tmpPath;
focal_class.type: Path
focal_class.var_name: tmpPath
focal_class.declarator: tmpJobInputPath
focal_class.modifier: private
focal_class.original_string: private Path tmpJobInputPath;
focal_class.type: Path
focal_class.var_name: tmpJobInputPath
focal_class.declarator: tmpJobOutputPath
focal_class.modifier: private
focal_class.original_string: private Path tmpJobOutputPath;
focal_class.type: Path
focal_class.var_name: tmpJobOutputPath
focal_class.declarator: FILES_TO_KEEP = 6
focal_class.modifier: private final
focal_class.original_string: private final int FILES_TO_KEEP = 6;
focal_class.type: int
focal_class.var_name: FILES_TO_KEEP
focal_class.file: databus-worker/src/main/java/com/inmobi/databus/local/LocalStreamService.java
focal_class.identifier: LocalStreamService
focal_class.class_method_signature: LocalStreamService.LocalStreamService(DatabusConfig config, Cluster cluster,
                              CheckpointProvider provider)
focal_class.full_signature: public  LocalStreamService(DatabusConfig config, Cluster cluster,
                              CheckpointProvider provider)
focal_class.identifier: LocalStreamService
focal_class.modifiers: public
focal_class.parameters: (DatabusConfig config, Cluster cluster,
                              CheckpointProvider provider)
focal_class.signature:  LocalStreamService(DatabusConfig config, Cluster cluster,
                              CheckpointProvider provider)
focal_class.class_method_signature: LocalStreamService.cleanUpTmp(FileSystem fs)
focal_class.full_signature: private void cleanUpTmp(FileSystem fs)
focal_class.identifier: cleanUpTmp
focal_class.modifiers: private
focal_class.parameters: (FileSystem fs)
focal_class.return: void
focal_class.signature: void cleanUpTmp(FileSystem fs)
focal_class.class_method_signature: LocalStreamService.getMSecondsTillNextRun(long currentTime)
focal_class.full_signature: @Override public long getMSecondsTillNextRun(long currentTime)
focal_class.identifier: getMSecondsTillNextRun
focal_class.modifiers: @Override public
focal_class.parameters: (long currentTime)
focal_class.return: long
focal_class.signature: long getMSecondsTillNextRun(long currentTime)
focal_class.class_method_signature: LocalStreamService.execute()
focal_class.full_signature: @Override protected void execute()
focal_class.identifier: execute
focal_class.modifiers: @Override protected
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void execute()
focal_class.class_method_signature: LocalStreamService.checkPoint(Map<String, FileStatus> checkPointPaths)
focal_class.full_signature: private void checkPoint(Map<String, FileStatus> checkPointPaths)
focal_class.identifier: checkPoint
focal_class.modifiers: private
focal_class.parameters: (Map<String, FileStatus> checkPointPaths)
focal_class.return: void
focal_class.signature: void checkPoint(Map<String, FileStatus> checkPointPaths)
focal_class.class_method_signature: LocalStreamService.prepareForCommit(long commitTime,
                                             Map<FileStatus, String> fileListing)
focal_class.full_signature: private Map<Path, Path> prepareForCommit(long commitTime,
                                             Map<FileStatus, String> fileListing)
focal_class.identifier: prepareForCommit
focal_class.modifiers: private
focal_class.parameters: (long commitTime,
                                             Map<FileStatus, String> fileListing)
focal_class.return: Map<Path, Path>
focal_class.signature: Map<Path, Path> prepareForCommit(long commitTime,
                                             Map<FileStatus, String> fileListing)
focal_class.class_method_signature: LocalStreamService.populateTrashCommitPaths(Set<FileStatus> trashSet)
focal_class.full_signature:  Map<Path, Path> populateTrashCommitPaths(Set<FileStatus> trashSet)
focal_class.identifier: populateTrashCommitPaths
focal_class.parameters: (Set<FileStatus> trashSet)
focal_class.return: Map<Path, Path>
focal_class.signature: Map<Path, Path> populateTrashCommitPaths(Set<FileStatus> trashSet)
focal_class.class_method_signature: LocalStreamService.commit(Map<Path, Path> commitPaths)
focal_class.full_signature: private void commit(Map<Path, Path> commitPaths)
focal_class.identifier: commit
focal_class.modifiers: private
focal_class.parameters: (Map<Path, Path> commitPaths)
focal_class.return: void
focal_class.signature: void commit(Map<Path, Path> commitPaths)
focal_class.class_method_signature: LocalStreamService.createMRInput(Path inputPath,
                               Map<FileStatus, String> fileListing, Set<FileStatus> trashSet,
                               Map<String, FileStatus> checkpointPaths)
focal_class.full_signature: private void createMRInput(Path inputPath,
                               Map<FileStatus, String> fileListing, Set<FileStatus> trashSet,
                               Map<String, FileStatus> checkpointPaths)
focal_class.identifier: createMRInput
focal_class.modifiers: private
focal_class.parameters: (Path inputPath,
                               Map<FileStatus, String> fileListing, Set<FileStatus> trashSet,
                               Map<String, FileStatus> checkpointPaths)
focal_class.return: void
focal_class.signature: void createMRInput(Path inputPath,
                               Map<FileStatus, String> fileListing, Set<FileStatus> trashSet,
                               Map<String, FileStatus> checkpointPaths)
focal_class.class_method_signature: LocalStreamService.createListing(FileSystem fs, FileStatus fileStatus,
                              Map<FileStatus, String> results, Set<FileStatus> trashSet,
                              Map<String, FileStatus> checkpointPaths)
focal_class.full_signature: public void createListing(FileSystem fs, FileStatus fileStatus,
                              Map<FileStatus, String> results, Set<FileStatus> trashSet,
                              Map<String, FileStatus> checkpointPaths)
focal_class.identifier: createListing
focal_class.modifiers: public
focal_class.parameters: (FileSystem fs, FileStatus fileStatus,
                              Map<FileStatus, String> results, Set<FileStatus> trashSet,
                              Map<String, FileStatus> checkpointPaths)
focal_class.return: void
focal_class.signature: void createListing(FileSystem fs, FileStatus fileStatus,
                              Map<FileStatus, String> results, Set<FileStatus> trashSet,
                              Map<String, FileStatus> checkpointPaths)
focal_class.class_method_signature: LocalStreamService.createListing(FileSystem fs, FileStatus fileStatus,
                              Map<FileStatus, String> results, Set<FileStatus> trashSet,
                              Map<String, FileStatus> checkpointPaths,
                              long lastFileTimeout)
focal_class.full_signature: public void createListing(FileSystem fs, FileStatus fileStatus,
                              Map<FileStatus, String> results, Set<FileStatus> trashSet,
                              Map<String, FileStatus> checkpointPaths,
                              long lastFileTimeout)
focal_class.identifier: createListing
focal_class.modifiers: public
focal_class.parameters: (FileSystem fs, FileStatus fileStatus,
                              Map<FileStatus, String> results, Set<FileStatus> trashSet,
                              Map<String, FileStatus> checkpointPaths,
                              long lastFileTimeout)
focal_class.return: void
focal_class.signature: void createListing(FileSystem fs, FileStatus fileStatus,
                              Map<FileStatus, String> results, Set<FileStatus> trashSet,
                              Map<String, FileStatus> checkpointPaths,
                              long lastFileTimeout)
focal_class.class_method_signature: LocalStreamService.processFile(FileStatus file, String currentFile,
                             String checkPointValue, FileSystem fs, Map<FileStatus, String> results,
                             Map<String, FileStatus> collectorPaths)
focal_class.full_signature: private void processFile(FileStatus file, String currentFile,
                             String checkPointValue, FileSystem fs, Map<FileStatus, String> results,
                             Map<String, FileStatus> collectorPaths)
focal_class.identifier: processFile
focal_class.modifiers: private
focal_class.parameters: (FileStatus file, String currentFile,
                             String checkPointValue, FileSystem fs, Map<FileStatus, String> results,
                             Map<String, FileStatus> collectorPaths)
focal_class.return: void
focal_class.signature: void processFile(FileStatus file, String currentFile,
                             String checkPointValue, FileSystem fs, Map<FileStatus, String> results,
                             Map<String, FileStatus> collectorPaths)
focal_class.class_method_signature: LocalStreamService.isEmptyFile(FileStatus fileStatus, FileSystem fs)
focal_class.full_signature: private boolean isEmptyFile(FileStatus fileStatus, FileSystem fs)
focal_class.identifier: isEmptyFile
focal_class.modifiers: private
focal_class.parameters: (FileStatus fileStatus, FileSystem fs)
focal_class.return: boolean
focal_class.signature: boolean isEmptyFile(FileStatus fileStatus, FileSystem fs)
focal_class.class_method_signature: LocalStreamService.populateCheckpointPathForCollector(
    Map<String, FileStatus> checkpointPaths,
    TreeMap<String, FileStatus> collectorPaths, String checkpointKey)
focal_class.full_signature: private void populateCheckpointPathForCollector(
    Map<String, FileStatus> checkpointPaths,
    TreeMap<String, FileStatus> collectorPaths, String checkpointKey)
focal_class.identifier: populateCheckpointPathForCollector
focal_class.modifiers: private
focal_class.parameters: (
    Map<String, FileStatus> checkpointPaths,
    TreeMap<String, FileStatus> collectorPaths, String checkpointKey)
focal_class.return: void
focal_class.signature: void populateCheckpointPathForCollector(
    Map<String, FileStatus> checkpointPaths,
    TreeMap<String, FileStatus> collectorPaths, String checkpointKey)
focal_class.class_method_signature: LocalStreamService.populateTrash(Map<String, FileStatus> collectorPaths,
                               Set<FileStatus> trashSet)
focal_class.full_signature: private void populateTrash(Map<String, FileStatus> collectorPaths,
                               Set<FileStatus> trashSet)
focal_class.identifier: populateTrash
focal_class.modifiers: private
focal_class.parameters: (Map<String, FileStatus> collectorPaths,
                               Set<FileStatus> trashSet)
focal_class.return: void
focal_class.signature: void populateTrash(Map<String, FileStatus> collectorPaths,
                               Set<FileStatus> trashSet)
focal_class.class_method_signature: LocalStreamService.aboveCheckpoint(String checkPoint, String file)
focal_class.full_signature: private boolean aboveCheckpoint(String checkPoint, String file)
focal_class.identifier: aboveCheckpoint
focal_class.modifiers: private
focal_class.parameters: (String checkPoint, String file)
focal_class.return: boolean
focal_class.signature: boolean aboveCheckpoint(String checkPoint, String file)
focal_class.class_method_signature: LocalStreamService.getCurrentFile(FileSystem fs, FileStatus[] files,
                                    long lastFileTimeout)
focal_class.full_signature: protected String getCurrentFile(FileSystem fs, FileStatus[] files,
                                    long lastFileTimeout)
focal_class.identifier: getCurrentFile
focal_class.modifiers: protected
focal_class.parameters: (FileSystem fs, FileStatus[] files,
                                    long lastFileTimeout)
focal_class.return: String
focal_class.signature: String getCurrentFile(FileSystem fs, FileStatus[] files,
                                    long lastFileTimeout)
focal_class.class_method_signature: LocalStreamService.getCategoryFromSrcPath(Path src)
focal_class.full_signature: private String getCategoryFromSrcPath(Path src)
focal_class.identifier: getCategoryFromSrcPath
focal_class.modifiers: private
focal_class.parameters: (Path src)
focal_class.return: String
focal_class.signature: String getCategoryFromSrcPath(Path src)
focal_class.class_method_signature: LocalStreamService.getCategoryFromDestPath(Path dest)
focal_class.full_signature: private String getCategoryFromDestPath(Path dest)
focal_class.identifier: getCategoryFromDestPath
focal_class.modifiers: private
focal_class.parameters: (Path dest)
focal_class.return: String
focal_class.signature: String getCategoryFromDestPath(Path dest)
focal_class.class_method_signature: LocalStreamService.getCategoryJobOutTmpPath(String category)
focal_class.full_signature: private Path getCategoryJobOutTmpPath(String category)
focal_class.identifier: getCategoryJobOutTmpPath
focal_class.modifiers: private
focal_class.parameters: (String category)
focal_class.return: Path
focal_class.signature: Path getCategoryJobOutTmpPath(String category)
focal_class.class_method_signature: LocalStreamService.createJob(Path inputPath)
focal_class.full_signature: private Job createJob(Path inputPath)
focal_class.identifier: createJob
focal_class.modifiers: private
focal_class.parameters: (Path inputPath)
focal_class.return: Job
focal_class.signature: Job createJob(Path inputPath)
focal_class.superclass: extends AbstractService
focal_method.body: Map<Path, Path> populateTrashCommitPaths(Set<FileStatus> trashSet) {
      // find trash paths
      Map<Path, Path> trashPaths = new TreeMap<Path, Path>();
      Path trash = cluster.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;
    }
focal_method.class_method_signature: LocalStreamService.populateTrashCommitPaths(Set<FileStatus> trashSet)
focal_method.full_signature:  Map<Path, Path> populateTrashCommitPaths(Set<FileStatus> trashSet)
focal_method.identifier: populateTrashCommitPaths
focal_method.invocations: getTrashPathWithDateHour iterator hasNext next getName getParent getPath getName getPath debug getPath put getPath
focal_method.parameters: (Set<FileStatus> trashSet)
focal_method.return: Map<Path, Path>
focal_method.signature: Map<Path, Path> populateTrashCommitPaths(Set<FileStatus> trashSet)