@Test public void testDeleteOnSource() throws Exception { State state = getTestState("testDeleteOnSource"); Path testMethodTempPath = new Path(testClassTempPath, "testDeleteOnSource"); DeletingCopyDataPublisher copyDataPublisher = closer.register(new DeletingCopyDataPublisher(state)); File outputDir = new File(testMethodTempPath.toString(), "task-output/jobid/1f042f494d1fe2198e0e71a17faa233f33b5099b"); outputDir.mkdirs(); outputDir.deleteOnExit(); WorkUnitState wus = new WorkUnitState(); CopyableDataset copyableDataset = new TestCopyableDataset(new Path("origin")); CopyableDatasetMetadata metadata = new CopyableDatasetMetadata(copyableDataset); CopyEntity cf = CopyableFileUtils.createTestCopyableFile(new Path(testMethodTempPath, "test.txt").toString()); CopySource.serializeCopyableDataset(wus, metadata); CopySource.serializeCopyEntity(wus, cf); Assert.assertTrue(fs.exists(new Path(testMethodTempPath, "test.txt"))); wus.setWorkingState(WorkingState.SUCCESSFUL); copyDataPublisher.publishData(ImmutableList.of(wus)); Assert.assertFalse(fs.exists(new Path(testMethodTempPath, "test.txt"))); }
@Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { super.publishData(states); for (WorkUnitState state : states) { if (state.getWorkingState() == WorkingState.COMMITTED) { try { deleteFilesOnSource(state); } catch (Throwable t) { log.warn( String.format("Failed to delete one or more files on source in %s", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE)), t); } } else { log.info(String.format("Not deleting files %s on source fileSystem as the workunit state is %s.", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE), state.getWorkingState())); } } }
DeletingCopyDataPublisher extends CopyDataPublisher { @Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { super.publishData(states); for (WorkUnitState state : states) { if (state.getWorkingState() == WorkingState.COMMITTED) { try { deleteFilesOnSource(state); } catch (Throwable t) { log.warn( String.format("Failed to delete one or more files on source in %s", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE)), t); } } else { log.info(String.format("Not deleting files %s on source fileSystem as the workunit state is %s.", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE), state.getWorkingState())); } } } }
DeletingCopyDataPublisher extends CopyDataPublisher { @Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { super.publishData(states); for (WorkUnitState state : states) { if (state.getWorkingState() == WorkingState.COMMITTED) { try { deleteFilesOnSource(state); } catch (Throwable t) { log.warn( String.format("Failed to delete one or more files on source in %s", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE)), t); } } else { log.info(String.format("Not deleting files %s on source fileSystem as the workunit state is %s.", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE), state.getWorkingState())); } } } DeletingCopyDataPublisher(State state); }
DeletingCopyDataPublisher extends CopyDataPublisher { @Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { super.publishData(states); for (WorkUnitState state : states) { if (state.getWorkingState() == WorkingState.COMMITTED) { try { deleteFilesOnSource(state); } catch (Throwable t) { log.warn( String.format("Failed to delete one or more files on source in %s", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE)), t); } } else { log.info(String.format("Not deleting files %s on source fileSystem as the workunit state is %s.", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE), state.getWorkingState())); } } } DeletingCopyDataPublisher(State state); @Override void publishData(Collection<? extends WorkUnitState> states); }
DeletingCopyDataPublisher extends CopyDataPublisher { @Override public void publishData(Collection<? extends WorkUnitState> states) throws IOException { super.publishData(states); for (WorkUnitState state : states) { if (state.getWorkingState() == WorkingState.COMMITTED) { try { deleteFilesOnSource(state); } catch (Throwable t) { log.warn( String.format("Failed to delete one or more files on source in %s", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE)), t); } } else { log.info(String.format("Not deleting files %s on source fileSystem as the workunit state is %s.", state.getProp(CopySource.SERIALIZED_COPYABLE_FILE), state.getWorkingState())); } } } DeletingCopyDataPublisher(State state); @Override void publishData(Collection<? extends WorkUnitState> states);  }