Skip to content

Commit

Permalink
NIFI-14158: DeleteHDFS processor ignores delete action return value
Browse files Browse the repository at this point in the history
  • Loading branch information
@mark-bathori
mark-bathori committed Jan 15, 2025
1 parent fd327e6 commit 931965f
Show file tree
Hide file tree
Showing 2 changed files with 90 additions and 6 deletions.
18 changes: 13 additions & 5 deletions ...ndle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/DeleteHDFS.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -172,11 +172,19 @@ public void onTrigger(ProcessContext context, ProcessSession session) throws Pro
attributes.put(getAttributePrefix() + ".path", path.getParent().toString());
flowFile = session.putAllAttributes(flowFile, attributes);

fileSystem.delete(path, isRecursive(context, session));
getLogger().debug("For flowfile {} Deleted file at path {} with name {}", originalFlowFile, path.getParent(), path.getName());
final Path qualifiedPath = path.makeQualified(fileSystem.getUri(), fileSystem.getWorkingDirectory());
flowFile = session.putAttribute(flowFile, HADOOP_FILE_URL_ATTRIBUTE, qualifiedPath.toString());
session.getProvenanceReporter().invokeRemoteProcess(flowFile, qualifiedPath.toString());
boolean success = fileSystem.delete(path, isRecursive(context, session));

if (success) {
getLogger().debug("For flowfile {} Deleted file at path {} with name {}", originalFlowFile, path.getParent(), path.getName());
final Path qualifiedPath = path.makeQualified(fileSystem.getUri(), fileSystem.getWorkingDirectory());
flowFile = session.putAttribute(flowFile, HADOOP_FILE_URL_ATTRIBUTE, qualifiedPath.toString());
session.getProvenanceReporter().invokeRemoteProcess(flowFile, qualifiedPath.toString());
} else {
getLogger().warn("Failed to delete file at path {} with name {} due to unknown issue, please check related component logs.", path.getParent(), path.getName());
attributes.put(getAttributePrefix() + ".error.message", "Delete action failed due to unknown issue, please check related component logs.");
session.transfer(session.putAllAttributes(session.clone(flowFile), attributes), getFailureRelationship());
failedPath++;
}
} catch (IOException ioe) {
if (handleAuthErrors(ioe, session, context, new GSSExceptionRollbackYieldSessionHandler())) {
return null;
Expand Down
78 changes: 77 additions & 1 deletion .../nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestDeleteHDFS.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
Expand All @@ -56,6 +57,7 @@ public void testSuccessfulDelete() throws Exception {
Path filePath = new Path("/some/path/to/file.txt");
when(mockFileSystem.exists(any(Path.class))).thenReturn(true);
when(mockFileSystem.getUri()).thenReturn(new URI("hdfs://0.example.com:8020"));
when(mockFileSystem.delete(any(Path.class), anyBoolean())).thenReturn(true);
DeleteHDFS deleteHDFS = new TestableDeleteHDFS(mockFileSystem);
TestRunner runner = TestRunners.newTestRunner(deleteHDFS);
runner.setIncomingConnection(false);
Expand All @@ -82,6 +84,7 @@ public void testDeleteFromIncomingFlowFile() throws Exception {
Path filePath = new Path("/some/path/to/file.txt");
when(mockFileSystem.exists(any(Path.class))).thenReturn(true);
when(mockFileSystem.getUri()).thenReturn(new URI("hdfs://0.example.com:8020"));
when(mockFileSystem.delete(any(Path.class), anyBoolean())).thenReturn(true);
DeleteHDFS deleteHDFS = new TestableDeleteHDFS(mockFileSystem);
TestRunner runner = TestRunners.newTestRunner(deleteHDFS);
runner.setProperty(DeleteHDFS.FILE_OR_DIRECTORY, "${hdfs.file}");
Expand Down Expand Up @@ -156,7 +159,7 @@ public void testNoFlowFilesWithIncomingConnection() {
}

@Test
public void testUnsuccessfulDelete() throws Exception {
public void testDeleteNotExistingFile() throws Exception {
Path filePath = new Path("/some/path/to/file.txt");
when(mockFileSystem.exists(any(Path.class))).thenReturn(false);
DeleteHDFS deleteHDFS = new TestableDeleteHDFS(mockFileSystem);
Expand All @@ -169,6 +172,22 @@ public void testUnsuccessfulDelete() throws Exception {
runner.assertTransferCount(DeleteHDFS.REL_FAILURE, 0);
}

@Test
public void testFailedDelete() throws Exception {
final Path filePath = new Path("/some/path/to/file.txt");
when(mockFileSystem.exists(any(Path.class))).thenReturn(true);
when(mockFileSystem.delete(any(Path.class), anyBoolean())).thenReturn(false);
final DeleteHDFS deleteHDFS = new TestableDeleteHDFS(mockFileSystem);
final TestRunner runner = TestRunners.newTestRunner(deleteHDFS);
runner.setIncomingConnection(false);
runner.assertNotValid();
runner.setProperty(DeleteHDFS.FILE_OR_DIRECTORY, filePath.toString());
runner.assertValid();
runner.run();
runner.assertTransferCount(DeleteHDFS.REL_SUCCESS, 0);
runner.assertTransferCount(DeleteHDFS.REL_FAILURE, 1);
}

@Test
public void testGlobDelete() throws Exception {
Path glob = new Path("/data/for/2017/08/05/*");
Expand All @@ -183,6 +202,7 @@ public void testGlobDelete() throws Exception {
when(mockFileSystem.exists(any(Path.class))).thenReturn(true);
when(mockFileSystem.globStatus(any(Path.class))).thenReturn(fileStatuses);
when(mockFileSystem.getUri()).thenReturn(new URI("hdfs://0.example.com:8020"));
when(mockFileSystem.delete(any(Path.class), anyBoolean())).thenReturn(true);
DeleteHDFS deleteHDFS = new TestableDeleteHDFS(mockFileSystem);
TestRunner runner = TestRunners.newTestRunner(deleteHDFS);
runner.setIncomingConnection(false);
Expand All @@ -193,6 +213,61 @@ public void testGlobDelete() throws Exception {
runner.assertTransferCount(DeleteHDFS.REL_SUCCESS, 1);
}

@Test
public void testFailedGlobDelete() throws Exception {
final Path glob = new Path("/data/for/2017/08/05/*");
final int fileCount = 10;
final FileStatus[] fileStatuses = new FileStatus[fileCount];
for (int i = 0; i < fileCount; i++) {
final Path file = new Path("/data/for/2017/08/05/file" + i);
final FileStatus fileStatus = mock(FileStatus.class);
when(fileStatus.getPath()).thenReturn(file);
fileStatuses[i] = fileStatus;
}
when(mockFileSystem.exists(any(Path.class))).thenReturn(true);
when(mockFileSystem.globStatus(any(Path.class))).thenReturn(fileStatuses);
when(mockFileSystem.getUri()).thenReturn(new URI("hdfs://0.example.com:8020"));
when(mockFileSystem.delete(any(Path.class), anyBoolean())).thenReturn(false);
final DeleteHDFS deleteHDFS = new TestableDeleteHDFS(mockFileSystem);
final TestRunner runner = TestRunners.newTestRunner(deleteHDFS);
runner.setIncomingConnection(false);
runner.assertNotValid();
runner.setProperty(DeleteHDFS.FILE_OR_DIRECTORY, glob.toString());
runner.assertValid();
runner.run();
runner.assertTransferCount(DeleteHDFS.REL_SUCCESS, 0);
runner.assertTransferCount(DeleteHDFS.REL_FAILURE, fileCount);
}

@Test
public void testMixedGlobDelete() throws Exception {
final Path glob = new Path("/data/for/2017/08/05/*");
final int fileCount = 3;
final FileStatus[] fileStatuses = new FileStatus[fileCount];
for (int i = 0; i < fileCount; i++) {
final Path file = new Path("/data/for/2017/08/05/file" + i);
final FileStatus fileStatus = mock(FileStatus.class);
when(fileStatus.getPath()).thenReturn(file);
fileStatuses[i] = fileStatus;
}
when(mockFileSystem.exists(any(Path.class))).thenReturn(true);
when(mockFileSystem.globStatus(any(Path.class))).thenReturn(fileStatuses);
when(mockFileSystem.getUri()).thenReturn(new URI("hdfs://0.example.com:8020"));
when(mockFileSystem.delete(any(Path.class), anyBoolean()))
.thenReturn(false)
.thenReturn(true)
.thenReturn(false);
final DeleteHDFS deleteHDFS = new TestableDeleteHDFS(mockFileSystem);
final TestRunner runner = TestRunners.newTestRunner(deleteHDFS);
runner.setIncomingConnection(false);
runner.assertNotValid();
runner.setProperty(DeleteHDFS.FILE_OR_DIRECTORY, glob.toString());
runner.assertValid();
runner.run();
runner.assertTransferCount(DeleteHDFS.REL_SUCCESS, 0);
runner.assertTransferCount(DeleteHDFS.REL_FAILURE, 2);
}

@Test
public void testGlobDeleteFromIncomingFlowFile() throws Exception {
Path glob = new Path("/data/for/2017/08/05/*");
Expand All @@ -207,6 +282,7 @@ public void testGlobDeleteFromIncomingFlowFile() throws Exception {
when(mockFileSystem.exists(any(Path.class))).thenReturn(true);
when(mockFileSystem.globStatus(any(Path.class))).thenReturn(fileStatuses);
when(mockFileSystem.getUri()).thenReturn(new URI("hdfs://0.example.com:8020"));
when(mockFileSystem.delete(any(Path.class), anyBoolean())).thenReturn(true);
DeleteHDFS deleteHDFS = new TestableDeleteHDFS(mockFileSystem);
TestRunner runner = TestRunners.newTestRunner(deleteHDFS);
runner.setIncomingConnection(true);
Expand Down

0 comments on commit 931965f

Please sign in to comment.