@Test(priority = 1) public void testPositive() throws Exception { Map<Path, FileSystem> consumePaths = new HashMap<Path, FileSystem>(); Path p = service.getDistCPInputFile(consumePaths, testRoot); LOG.info("distcp input [" + p + "]"); FSDataInputStream in = localFs.open(p); BufferedReader reader = new BufferedReader(new InputStreamReader(in)); String result = reader.readLine(); assert result.equals(expectedFileName1); result = reader.readLine(); assert result.equals(expectedFileName2); reader.close(); Set<String> resultSet = new HashSet<String>(); for (Path consumePath : consumePaths.keySet()) { if (!consumePath.toString().contains("file:/tmp/com.inmobi.databus" + ".distcp.TestDistCPBaseService/databusCluster")) { LOG.info("Path to consume [" + consumePath + "]"); resultSet.add(consumePath.toString()); } } assert resultSet.containsAll(expectedConsumePaths); assert consumePaths.size() == resultSet.size() + 1; }
protected Path getDistCPInputFile(Map<Path, FileSystem> consumePaths, Path tmp) throws IOException { Path input = getInputPath(); if (!srcFs.exists(input)) return null; FileStatus[] fileList = srcFs.listStatus(input); if (fileList != null) { Set<String> minFilesSet = new HashSet<String>(); for (int i = 0; i < fileList.length; i++) { Path consumeFilePath = fileList[i].getPath().makeQualified(srcFs); consumePaths.put(consumeFilePath, srcFs); LOG.debug("Reading minutePaths from ConsumePath [" + consumeFilePath + "]"); readConsumePath(srcFs, consumeFilePath, minFilesSet); } Path tmpPath = createInputFileForDISCTP(destFs, srcCluster.getName(), tmp, minFilesSet); return getFinalPathForDistCP(tmpPath, consumePaths); } return null; }
DistcpBaseService extends AbstractService { protected Path getDistCPInputFile(Map<Path, FileSystem> consumePaths, Path tmp) throws IOException { Path input = getInputPath(); if (!srcFs.exists(input)) return null; FileStatus[] fileList = srcFs.listStatus(input); if (fileList != null) { Set<String> minFilesSet = new HashSet<String>(); for (int i = 0; i < fileList.length; i++) { Path consumeFilePath = fileList[i].getPath().makeQualified(srcFs); consumePaths.put(consumeFilePath, srcFs); LOG.debug("Reading minutePaths from ConsumePath [" + consumeFilePath + "]"); readConsumePath(srcFs, consumeFilePath, minFilesSet); } Path tmpPath = createInputFileForDISCTP(destFs, srcCluster.getName(), tmp, minFilesSet); return getFinalPathForDistCP(tmpPath, consumePaths); } return null; } }
DistcpBaseService extends AbstractService { protected Path getDistCPInputFile(Map<Path, FileSystem> consumePaths, Path tmp) throws IOException { Path input = getInputPath(); if (!srcFs.exists(input)) return null; FileStatus[] fileList = srcFs.listStatus(input); if (fileList != null) { Set<String> minFilesSet = new HashSet<String>(); for (int i = 0; i < fileList.length; i++) { Path consumeFilePath = fileList[i].getPath().makeQualified(srcFs); consumePaths.put(consumeFilePath, srcFs); LOG.debug("Reading minutePaths from ConsumePath [" + consumeFilePath + "]"); readConsumePath(srcFs, consumeFilePath, minFilesSet); } Path tmpPath = createInputFileForDISCTP(destFs, srcCluster.getName(), tmp, minFilesSet); return getFinalPathForDistCP(tmpPath, consumePaths); } return null; } DistcpBaseService(DatabusConfig config, String name,
                           Cluster srcCluster,
                           Cluster destCluster); }
DistcpBaseService extends AbstractService { protected Path getDistCPInputFile(Map<Path, FileSystem> consumePaths, Path tmp) throws IOException { Path input = getInputPath(); if (!srcFs.exists(input)) return null; FileStatus[] fileList = srcFs.listStatus(input); if (fileList != null) { Set<String> minFilesSet = new HashSet<String>(); for (int i = 0; i < fileList.length; i++) { Path consumeFilePath = fileList[i].getPath().makeQualified(srcFs); consumePaths.put(consumeFilePath, srcFs); LOG.debug("Reading minutePaths from ConsumePath [" + consumeFilePath + "]"); readConsumePath(srcFs, consumeFilePath, minFilesSet); } Path tmpPath = createInputFileForDISCTP(destFs, srcCluster.getName(), tmp, minFilesSet); return getFinalPathForDistCP(tmpPath, consumePaths); } return null; } DistcpBaseService(DatabusConfig config, String name,
                           Cluster srcCluster,
                           Cluster destCluster); @Override long getMSecondsTillNextRun(long currentTime); }
DistcpBaseService extends AbstractService { protected Path getDistCPInputFile(Map<Path, FileSystem> consumePaths, Path tmp) throws IOException { Path input = getInputPath(); if (!srcFs.exists(input)) return null; FileStatus[] fileList = srcFs.listStatus(input); if (fileList != null) { Set<String> minFilesSet = new HashSet<String>(); for (int i = 0; i < fileList.length; i++) { Path consumeFilePath = fileList[i].getPath().makeQualified(srcFs); consumePaths.put(consumeFilePath, srcFs); LOG.debug("Reading minutePaths from ConsumePath [" + consumeFilePath + "]"); readConsumePath(srcFs, consumeFilePath, minFilesSet); } Path tmpPath = createInputFileForDISCTP(destFs, srcCluster.getName(), tmp, minFilesSet); return getFinalPathForDistCP(tmpPath, consumePaths); } return null; } DistcpBaseService(DatabusConfig config, String name,
                           Cluster srcCluster,
                           Cluster destCluster); @Override long getMSecondsTillNextRun(long currentTime);  }