@Test public void testGetFlowFilesNoUgiShouldntCallDoAs() throws Exception { hdfsResources = new AbstractHadoopProcessor.HdfsResources(configuration, fileSystem, null); init(); SequenceFileReader reader = mock(SequenceFileReader.class); Path file = mock(Path.class); getHDFSSequenceFile.getFlowFiles(configuration, fileSystem, reader, file); verify(reader).readSequenceFile(file, configuration, fileSystem); }
protected Set<FlowFile> getFlowFiles(final Configuration conf, final FileSystem hdfs, final SequenceFileReader<Set<FlowFile>> reader, final Path file) throws Exception { PrivilegedExceptionAction<Set<FlowFile>> privilegedExceptionAction = new PrivilegedExceptionAction<Set<FlowFile>>() { @Override public Set<FlowFile> run() throws Exception { return reader.readSequenceFile(file, conf, hdfs); } }; UserGroupInformation userGroupInformation = getUserGroupInformation(); if (userGroupInformation == null) { return privilegedExceptionAction.run(); } else { return userGroupInformation.doAs(privilegedExceptionAction); } }
GetHDFSSequenceFile extends GetHDFS { protected Set<FlowFile> getFlowFiles(final Configuration conf, final FileSystem hdfs, final SequenceFileReader<Set<FlowFile>> reader, final Path file) throws Exception { PrivilegedExceptionAction<Set<FlowFile>> privilegedExceptionAction = new PrivilegedExceptionAction<Set<FlowFile>>() { @Override public Set<FlowFile> run() throws Exception { return reader.readSequenceFile(file, conf, hdfs); } }; UserGroupInformation userGroupInformation = getUserGroupInformation(); if (userGroupInformation == null) { return privilegedExceptionAction.run(); } else { return userGroupInformation.doAs(privilegedExceptionAction); } } }
GetHDFSSequenceFile extends GetHDFS { protected Set<FlowFile> getFlowFiles(final Configuration conf, final FileSystem hdfs, final SequenceFileReader<Set<FlowFile>> reader, final Path file) throws Exception { PrivilegedExceptionAction<Set<FlowFile>> privilegedExceptionAction = new PrivilegedExceptionAction<Set<FlowFile>>() { @Override public Set<FlowFile> run() throws Exception { return reader.readSequenceFile(file, conf, hdfs); } }; UserGroupInformation userGroupInformation = getUserGroupInformation(); if (userGroupInformation == null) { return privilegedExceptionAction.run(); } else { return userGroupInformation.doAs(privilegedExceptionAction); } }  }
GetHDFSSequenceFile extends GetHDFS { protected Set<FlowFile> getFlowFiles(final Configuration conf, final FileSystem hdfs, final SequenceFileReader<Set<FlowFile>> reader, final Path file) throws Exception { PrivilegedExceptionAction<Set<FlowFile>> privilegedExceptionAction = new PrivilegedExceptionAction<Set<FlowFile>>() { @Override public Set<FlowFile> run() throws Exception { return reader.readSequenceFile(file, conf, hdfs); } }; UserGroupInformation userGroupInformation = getUserGroupInformation(); if (userGroupInformation == null) { return privilegedExceptionAction.run(); } else { return userGroupInformation.doAs(privilegedExceptionAction); } }   }
GetHDFSSequenceFile extends GetHDFS { protected Set<FlowFile> getFlowFiles(final Configuration conf, final FileSystem hdfs, final SequenceFileReader<Set<FlowFile>> reader, final Path file) throws Exception { PrivilegedExceptionAction<Set<FlowFile>> privilegedExceptionAction = new PrivilegedExceptionAction<Set<FlowFile>>() { @Override public Set<FlowFile> run() throws Exception { return reader.readSequenceFile(file, conf, hdfs); } }; UserGroupInformation userGroupInformation = getUserGroupInformation(); if (userGroupInformation == null) { return privilegedExceptionAction.run(); } else { return userGroupInformation.doAs(privilegedExceptionAction); } }    }