@Test @Ignore("Ignore it for we need to do some reactor work for storage cleanup job") public void test() throws Exception { FileSystem mockFs = mock(FileSystem.class); prepareUnusedIntermediateHiveTable(mockFs); prepareUnusedHDFSFiles(mockFs); MockStorageCleanupJob job = new MockStorageCleanupJob(KylinConfig.getInstanceFromEnv(), mockFs, mockFs); job.execute(new String[] { "--delete", "true" }); ArgumentCaptor<Path> pathCaptor = ArgumentCaptor.forClass(Path.class); verify(mockFs, times(2)).delete(pathCaptor.capture(), eq(true)); ArrayList<Path> expected = Lists.newArrayList( new Path("file: new Path("file: ); assertEquals(expected, pathCaptor.getAllValues()); }
@Override protected void execute(OptionsHelper optionsHelper) throws Exception { logger.info("options: '" + optionsHelper.getOptionsAsString() + "'"); logger.info("delete option value: '" + optionsHelper.getOptionValue(OPTION_DELETE) + "'"); logger.info("force option value: '" + optionsHelper.getOptionValue(OPTION_FORCE) + "'"); logger.info("thread option value: '" + optionsHelper.getOptionValue(THREAD_NUM) + "'"); delete = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_DELETE)); force = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_FORCE)); try { String threads = optionsHelper.getOptionValue(THREAD_NUM); if (threads != null) { threadsNum = Integer.parseInt(threads); } } catch (Exception e) { logger.info("Failed to parse value: {} for thread option: {}", optionsHelper.getOptionValue(THREAD_NUM), THREAD_NUM); } cleanup(); }
StorageCleanupJob extends AbstractApplication { @Override protected void execute(OptionsHelper optionsHelper) throws Exception { logger.info("options: '" + optionsHelper.getOptionsAsString() + "'"); logger.info("delete option value: '" + optionsHelper.getOptionValue(OPTION_DELETE) + "'"); logger.info("force option value: '" + optionsHelper.getOptionValue(OPTION_FORCE) + "'"); logger.info("thread option value: '" + optionsHelper.getOptionValue(THREAD_NUM) + "'"); delete = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_DELETE)); force = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_FORCE)); try { String threads = optionsHelper.getOptionValue(THREAD_NUM); if (threads != null) { threadsNum = Integer.parseInt(threads); } } catch (Exception e) { logger.info("Failed to parse value: {} for thread option: {}", optionsHelper.getOptionValue(THREAD_NUM), THREAD_NUM); } cleanup(); } }
StorageCleanupJob extends AbstractApplication { @Override protected void execute(OptionsHelper optionsHelper) throws Exception { logger.info("options: '" + optionsHelper.getOptionsAsString() + "'"); logger.info("delete option value: '" + optionsHelper.getOptionValue(OPTION_DELETE) + "'"); logger.info("force option value: '" + optionsHelper.getOptionValue(OPTION_FORCE) + "'"); logger.info("thread option value: '" + optionsHelper.getOptionValue(THREAD_NUM) + "'"); delete = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_DELETE)); force = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_FORCE)); try { String threads = optionsHelper.getOptionValue(THREAD_NUM); if (threads != null) { threadsNum = Integer.parseInt(threads); } } catch (Exception e) { logger.info("Failed to parse value: {} for thread option: {}", optionsHelper.getOptionValue(THREAD_NUM), THREAD_NUM); } cleanup(); } StorageCleanupJob(); protected  StorageCleanupJob(KylinConfig config, FileSystem defaultFs, FileSystem hbaseFs); }
StorageCleanupJob extends AbstractApplication { @Override protected void execute(OptionsHelper optionsHelper) throws Exception { logger.info("options: '" + optionsHelper.getOptionsAsString() + "'"); logger.info("delete option value: '" + optionsHelper.getOptionValue(OPTION_DELETE) + "'"); logger.info("force option value: '" + optionsHelper.getOptionValue(OPTION_FORCE) + "'"); logger.info("thread option value: '" + optionsHelper.getOptionValue(THREAD_NUM) + "'"); delete = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_DELETE)); force = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_FORCE)); try { String threads = optionsHelper.getOptionValue(THREAD_NUM); if (threads != null) { threadsNum = Integer.parseInt(threads); } } catch (Exception e) { logger.info("Failed to parse value: {} for thread option: {}", optionsHelper.getOptionValue(THREAD_NUM), THREAD_NUM); } cleanup(); } StorageCleanupJob(); protected  StorageCleanupJob(KylinConfig config, FileSystem defaultFs, FileSystem hbaseFs); void setDelete(boolean delete); void setForce(boolean force); List<String> getHiveGarbageTables(); List<String> getHbaseGarbageTables(); List<String> getHdfsGarbageFiles(); long getHdfsFileGarbageBytes(); void cleanup(); }
StorageCleanupJob extends AbstractApplication { @Override protected void execute(OptionsHelper optionsHelper) throws Exception { logger.info("options: '" + optionsHelper.getOptionsAsString() + "'"); logger.info("delete option value: '" + optionsHelper.getOptionValue(OPTION_DELETE) + "'"); logger.info("force option value: '" + optionsHelper.getOptionValue(OPTION_FORCE) + "'"); logger.info("thread option value: '" + optionsHelper.getOptionValue(THREAD_NUM) + "'"); delete = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_DELETE)); force = Boolean.parseBoolean(optionsHelper.getOptionValue(OPTION_FORCE)); try { String threads = optionsHelper.getOptionValue(THREAD_NUM); if (threads != null) { threadsNum = Integer.parseInt(threads); } } catch (Exception e) { logger.info("Failed to parse value: {} for thread option: {}", optionsHelper.getOptionValue(THREAD_NUM), THREAD_NUM); } cleanup(); } StorageCleanupJob(); protected  StorageCleanupJob(KylinConfig config, FileSystem defaultFs, FileSystem hbaseFs); void setDelete(boolean delete); void setForce(boolean force); List<String> getHiveGarbageTables(); List<String> getHbaseGarbageTables(); List<String> getHdfsGarbageFiles(); long getHdfsFileGarbageBytes(); void cleanup();  }