### Question:
BasicDeleteTaskExecutor implements TaskExecutor { @Override public void execute(TaskExecutionContext context, TaskInfo task) throws IOException, InterruptedException { DeleteTaskInfo delete = (DeleteTaskInfo) task; String resolved = TaskExecutors.resolvePath(context, delete.getPath()); LOG.info("delete file: {} ({})", resolved, delete.getPathKind()); switch (delete.getPathKind()) { case LOCAL_FILE_SYSTEM: deleteOnLocalFileSystem(context, resolved); break; case HADOOP_FILE_SYSTEM: deleteOnHadoopFileSystem(context, resolved); break; default: throw new AssertionError(delete.getPathKind()); } }  @Override boolean isSupported(TaskExecutionContext context, TaskInfo task); @Override void execute(TaskExecutionContext context, TaskInfo task);  }### Answer:
@Test public void simple() throws Exception { TaskExecutor executor = new BasicDeleteTaskExecutor(); temporary.newFile(); executor.execute(context, new BasicDeleteTaskInfo( "testing", PathKind.LOCAL_FILE_SYSTEM, temporary.getRoot().getAbsolutePath())); assertThat(temporary.getRoot().exists(), is(false)); }
@Test public void missing() throws Exception { TaskExecutor executor = new BasicDeleteTaskExecutor(); executor.execute(context, new BasicDeleteTaskInfo( "testing", PathKind.LOCAL_FILE_SYSTEM, new File(temporary.getRoot(), "__MISSING__").getAbsolutePath())); assertThat(temporary.getRoot().exists(), is(true)); }
@Test public void vars() throws Exception { TaskExecutor executor = new BasicDeleteTaskExecutor(); File batch = temporary.newFile(context.getBatchId()); File flow = temporary.newFile(context.getFlowId()); File exec = temporary.newFile(context.getExecutionId()); assertThat(batch.isFile(), is(true)); assertThat(flow.isFile(), is(true)); assertThat(exec.isFile(), is(true)); executor.execute(context, new BasicDeleteTaskInfo( "testing", PathKind.LOCAL_FILE_SYSTEM, new File(temporary.getRoot(), "${batch_id}").getAbsolutePath())); assertThat(batch.isFile(), is(false)); assertThat(flow.isFile(), is(true)); assertThat(exec.isFile(), is(true)); executor.execute(context, new BasicDeleteTaskInfo( "testing", PathKind.LOCAL_FILE_SYSTEM, new File(temporary.getRoot(), "${flow_id}").getAbsolutePath())); assertThat(batch.isFile(), is(false)); assertThat(flow.isFile(), is(false)); assertThat(exec.isFile(), is(true)); executor.execute(context, new BasicDeleteTaskInfo( "testing", PathKind.LOCAL_FILE_SYSTEM, new File(temporary.getRoot(), "${execution_id}").getAbsolutePath())); assertThat(batch.isFile(), is(false)); assertThat(flow.isFile(), is(false)); assertThat(exec.isFile(), is(false)); executor.execute(context, new BasicDeleteTaskInfo( "testing", PathKind.LOCAL_FILE_SYSTEM, new File(temporary.getRoot(), "${user}").getAbsolutePath())); }