@Test public void testCheckSrcLogFiles() throws Exception { createFileHelper(srcFs, new Path(srcLogDir, "testLog.log")); createFileHelper(srcFs, new Path(srcLogDir, "testLog.log.1")); this.logCopier.runOneIteration(); FileStatus[] destLogFiles = this.destFs.listStatus(destLogDir); Assert.assertEquals(destLogFiles.length, 1); Assert.assertEquals(destLogFiles[0].getLen(), testString.length() + 1); createFileHelper(srcFs, new Path(srcLogDir, "testLog.log.2")); this.logCopier.runOneIteration(); destLogFiles = this.destFs.listStatus(destLogDir); Assert.assertEquals(destLogFiles.length, 2); Assert.assertEquals(destLogFiles[0].getLen(), testString.length() + 1); Assert.assertEquals(destLogFiles[1].getLen(), testString.length() + 1); this.logCopier.shutDown(); destLogFiles = this.destFs.listStatus(destLogDir); Assert.assertEquals(destLogFiles.length, 3); Assert.assertEquals(destLogFiles[0].getLen(), testString.length() + 1); Assert.assertEquals(destLogFiles[1].getLen(), testString.length() + 1); Assert.assertEquals(destLogFiles[2].getLen(), testString.length() + 1); }
@VisibleForTesting void checkSrcLogFiles() throws IOException { List<FileStatus> srcLogFiles = new ArrayList<>(); Set<String> srcLogFileNames = new HashSet<>(); Set<Path> newLogFiles = new HashSet<>(); for (Path logDirPath : srcLogDirs) { srcLogFiles.addAll(FileListUtils.listFilesRecursively(srcFs, logDirPath)); for (FileStatus srcLogFile : srcLogFiles) { if (shouldIncludeLogFile(srcLogFile)) { newLogFiles.add(srcLogFile.getPath()); } srcLogFileNames.add(srcLogFile.getPath().getName()); } } if (newLogFiles.isEmpty()) { LOGGER.warn("No log file found under directories " + this.srcLogDirs); return; } List<Future> futures = new ArrayList<>(); for (final Path srcLogFile : newLogFiles) { String destLogFileName = this.logFileNamePrefix.isPresent() ? this.logFileNamePrefix.get() + "." + srcLogFile.getName() : srcLogFile.getName(); final Path destLogFile = new Path(this.destLogDir, destLogFileName); futures.add(this.executorService.submit(new LogCopyTask(srcLogFile, destLogFile))); } for (Future future : futures) { try { future.get(); } catch (InterruptedException e) { LOGGER.error("LogCopyTask was interrupted - {}", e); } catch (ExecutionException e) { LOGGER.error("Failed LogCopyTask - {}", e); } } if (needToUpdateDestFs) { if (destFsSupplier == null) { throw new IOException("Try to update dest fileSystem but destFsSupplier has not been set"); } this.destFs.close(); this.destFs = destFsSupplier.getFileSystem(); LOGGER.info("Dest fs updated" + destFs.toString()); needToUpdateDestFs = false; } if (needToUpdateSrcFs) { if (srcFsSupplier == null) { throw new IOException("Try to update source fileSystem but srcFsSupplier has not been set"); } this.srcFs.close(); this.srcFs = srcFsSupplier.getFileSystem(); LOGGER.info("Src fs updated" + srcFs.toString()); needToUpdateSrcFs = false; } pruneCopiedFileNames(srcLogFileNames); }
LogCopier extends AbstractScheduledService { @VisibleForTesting void checkSrcLogFiles() throws IOException { List<FileStatus> srcLogFiles = new ArrayList<>(); Set<String> srcLogFileNames = new HashSet<>(); Set<Path> newLogFiles = new HashSet<>(); for (Path logDirPath : srcLogDirs) { srcLogFiles.addAll(FileListUtils.listFilesRecursively(srcFs, logDirPath)); for (FileStatus srcLogFile : srcLogFiles) { if (shouldIncludeLogFile(srcLogFile)) { newLogFiles.add(srcLogFile.getPath()); } srcLogFileNames.add(srcLogFile.getPath().getName()); } } if (newLogFiles.isEmpty()) { LOGGER.warn("No log file found under directories " + this.srcLogDirs); return; } List<Future> futures = new ArrayList<>(); for (final Path srcLogFile : newLogFiles) { String destLogFileName = this.logFileNamePrefix.isPresent() ? this.logFileNamePrefix.get() + "." + srcLogFile.getName() : srcLogFile.getName(); final Path destLogFile = new Path(this.destLogDir, destLogFileName); futures.add(this.executorService.submit(new LogCopyTask(srcLogFile, destLogFile))); } for (Future future : futures) { try { future.get(); } catch (InterruptedException e) { LOGGER.error("LogCopyTask was interrupted - {}", e); } catch (ExecutionException e) { LOGGER.error("Failed LogCopyTask - {}", e); } } if (needToUpdateDestFs) { if (destFsSupplier == null) { throw new IOException("Try to update dest fileSystem but destFsSupplier has not been set"); } this.destFs.close(); this.destFs = destFsSupplier.getFileSystem(); LOGGER.info("Dest fs updated" + destFs.toString()); needToUpdateDestFs = false; } if (needToUpdateSrcFs) { if (srcFsSupplier == null) { throw new IOException("Try to update source fileSystem but srcFsSupplier has not been set"); } this.srcFs.close(); this.srcFs = srcFsSupplier.getFileSystem(); LOGGER.info("Src fs updated" + srcFs.toString()); needToUpdateSrcFs = false; } pruneCopiedFileNames(srcLogFileNames); } }
LogCopier extends AbstractScheduledService { @VisibleForTesting void checkSrcLogFiles() throws IOException { List<FileStatus> srcLogFiles = new ArrayList<>(); Set<String> srcLogFileNames = new HashSet<>(); Set<Path> newLogFiles = new HashSet<>(); for (Path logDirPath : srcLogDirs) { srcLogFiles.addAll(FileListUtils.listFilesRecursively(srcFs, logDirPath)); for (FileStatus srcLogFile : srcLogFiles) { if (shouldIncludeLogFile(srcLogFile)) { newLogFiles.add(srcLogFile.getPath()); } srcLogFileNames.add(srcLogFile.getPath().getName()); } } if (newLogFiles.isEmpty()) { LOGGER.warn("No log file found under directories " + this.srcLogDirs); return; } List<Future> futures = new ArrayList<>(); for (final Path srcLogFile : newLogFiles) { String destLogFileName = this.logFileNamePrefix.isPresent() ? this.logFileNamePrefix.get() + "." + srcLogFile.getName() : srcLogFile.getName(); final Path destLogFile = new Path(this.destLogDir, destLogFileName); futures.add(this.executorService.submit(new LogCopyTask(srcLogFile, destLogFile))); } for (Future future : futures) { try { future.get(); } catch (InterruptedException e) { LOGGER.error("LogCopyTask was interrupted - {}", e); } catch (ExecutionException e) { LOGGER.error("Failed LogCopyTask - {}", e); } } if (needToUpdateDestFs) { if (destFsSupplier == null) { throw new IOException("Try to update dest fileSystem but destFsSupplier has not been set"); } this.destFs.close(); this.destFs = destFsSupplier.getFileSystem(); LOGGER.info("Dest fs updated" + destFs.toString()); needToUpdateDestFs = false; } if (needToUpdateSrcFs) { if (srcFsSupplier == null) { throw new IOException("Try to update source fileSystem but srcFsSupplier has not been set"); } this.srcFs.close(); this.srcFs = srcFsSupplier.getFileSystem(); LOGGER.info("Src fs updated" + srcFs.toString()); needToUpdateSrcFs = false; } pruneCopiedFileNames(srcLogFileNames); } private  LogCopier(Builder builder); }
LogCopier extends AbstractScheduledService { @VisibleForTesting void checkSrcLogFiles() throws IOException { List<FileStatus> srcLogFiles = new ArrayList<>(); Set<String> srcLogFileNames = new HashSet<>(); Set<Path> newLogFiles = new HashSet<>(); for (Path logDirPath : srcLogDirs) { srcLogFiles.addAll(FileListUtils.listFilesRecursively(srcFs, logDirPath)); for (FileStatus srcLogFile : srcLogFiles) { if (shouldIncludeLogFile(srcLogFile)) { newLogFiles.add(srcLogFile.getPath()); } srcLogFileNames.add(srcLogFile.getPath().getName()); } } if (newLogFiles.isEmpty()) { LOGGER.warn("No log file found under directories " + this.srcLogDirs); return; } List<Future> futures = new ArrayList<>(); for (final Path srcLogFile : newLogFiles) { String destLogFileName = this.logFileNamePrefix.isPresent() ? this.logFileNamePrefix.get() + "." + srcLogFile.getName() : srcLogFile.getName(); final Path destLogFile = new Path(this.destLogDir, destLogFileName); futures.add(this.executorService.submit(new LogCopyTask(srcLogFile, destLogFile))); } for (Future future : futures) { try { future.get(); } catch (InterruptedException e) { LOGGER.error("LogCopyTask was interrupted - {}", e); } catch (ExecutionException e) { LOGGER.error("Failed LogCopyTask - {}", e); } } if (needToUpdateDestFs) { if (destFsSupplier == null) { throw new IOException("Try to update dest fileSystem but destFsSupplier has not been set"); } this.destFs.close(); this.destFs = destFsSupplier.getFileSystem(); LOGGER.info("Dest fs updated" + destFs.toString()); needToUpdateDestFs = false; } if (needToUpdateSrcFs) { if (srcFsSupplier == null) { throw new IOException("Try to update source fileSystem but srcFsSupplier has not been set"); } this.srcFs.close(); this.srcFs = srcFsSupplier.getFileSystem(); LOGGER.info("Src fs updated" + srcFs.toString()); needToUpdateSrcFs = false; } pruneCopiedFileNames(srcLogFileNames); } private  LogCopier(Builder builder); static Builder newBuilder(); }
LogCopier extends AbstractScheduledService { @VisibleForTesting void checkSrcLogFiles() throws IOException { List<FileStatus> srcLogFiles = new ArrayList<>(); Set<String> srcLogFileNames = new HashSet<>(); Set<Path> newLogFiles = new HashSet<>(); for (Path logDirPath : srcLogDirs) { srcLogFiles.addAll(FileListUtils.listFilesRecursively(srcFs, logDirPath)); for (FileStatus srcLogFile : srcLogFiles) { if (shouldIncludeLogFile(srcLogFile)) { newLogFiles.add(srcLogFile.getPath()); } srcLogFileNames.add(srcLogFile.getPath().getName()); } } if (newLogFiles.isEmpty()) { LOGGER.warn("No log file found under directories " + this.srcLogDirs); return; } List<Future> futures = new ArrayList<>(); for (final Path srcLogFile : newLogFiles) { String destLogFileName = this.logFileNamePrefix.isPresent() ? this.logFileNamePrefix.get() + "." + srcLogFile.getName() : srcLogFile.getName(); final Path destLogFile = new Path(this.destLogDir, destLogFileName); futures.add(this.executorService.submit(new LogCopyTask(srcLogFile, destLogFile))); } for (Future future : futures) { try { future.get(); } catch (InterruptedException e) { LOGGER.error("LogCopyTask was interrupted - {}", e); } catch (ExecutionException e) { LOGGER.error("Failed LogCopyTask - {}", e); } } if (needToUpdateDestFs) { if (destFsSupplier == null) { throw new IOException("Try to update dest fileSystem but destFsSupplier has not been set"); } this.destFs.close(); this.destFs = destFsSupplier.getFileSystem(); LOGGER.info("Dest fs updated" + destFs.toString()); needToUpdateDestFs = false; } if (needToUpdateSrcFs) { if (srcFsSupplier == null) { throw new IOException("Try to update source fileSystem but srcFsSupplier has not been set"); } this.srcFs.close(); this.srcFs = srcFsSupplier.getFileSystem(); LOGGER.info("Src fs updated" + srcFs.toString()); needToUpdateSrcFs = false; } pruneCopiedFileNames(srcLogFileNames); } private  LogCopier(Builder builder); static Builder newBuilder();  }