Use UGI doAs in DeleteHDFS
@@ -21,6 +21,7 @@ import com.google.common.collect.Maps;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.Restricted;
import org.apache.nifi.annotation.behavior.Restriction;
@@ -41,6 +42,7 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import java.io.IOException;
+import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
@@ -141,11 +143,15 @@ public class DeleteHDFS extends AbstractHadoopProcessor {
}
// We need a FlowFile to report provenance correctly.
- FlowFile flowFile = originalFlowFile != null ? originalFlowFile : session.create();
+ final FlowFile finalFlowFile = originalFlowFile != null ? originalFlowFile : session.create();
- final String fileOrDirectoryName = context.getProperty(FILE_OR_DIRECTORY).evaluateAttributeExpressions(flowFile).getValue();
+ final String fileOrDirectoryName = context.getProperty(FILE_OR_DIRECTORY).evaluateAttributeExpressions(finalFlowFile).getValue();
final FileSystem fileSystem = getFileSystem();
+ final UserGroupInformation ugi = getUserGroupInformation();
+
+ ugi.doAs((PrivilegedAction<Object>)() -> {
+ FlowFile flowFile = finalFlowFile;
try {
// Check if the user has supplied a file or directory pattern
List<Path> pathList = Lists.newArrayList();
@@ -199,5 +205,8 @@ public class DeleteHDFS extends AbstractHadoopProcessor {
session.transfer(flowFile, DeleteHDFS.REL_FAILURE);
}
+ return null;
+ });
+
}
}
\ No newline at end of file
