HDFSFileAppender extends AbstractAppender { public static void runOnHDFS(final UserGroupInformation userGroupInformation, final Configuration conf, final Consumer<FileSystem> func) { try { userGroupInformation.doAs(new PrivilegedExceptionAction<Void>() { @Override public Void run() { final FileSystem hdfs = getHDFS(conf); func.accept(hdfs); return null; } }); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); } catch (final IOException e) { throw new UncheckedIOException(e); } } @Inject  HDFSFileAppender(final ErrorReceiverProxy errorReceiverProxy,
                     final PathCreator pathCreator); static Configuration buildConfiguration(final String hdfsUri); static FileSystem getHDFS(final Configuration conf); static UserGroupInformation buildRemoteUser(final Optional<String> runAsUser); static void runOnHDFS(final UserGroupInformation userGroupInformation, final Configuration conf,
                                 final Consumer<FileSystem> func); @Override void startProcessing(); @PipelineProperty( description = "One or more destination paths for output files separated with commas. Replacement variables can be used in path strings such as ${feed}.", displayPriority = 1) void setOutputPaths(final String outputPaths); @PipelineProperty(description = "URI for the Hadoop Distributed File System (HDFS) to connect to, e.g. hdfs://mynamenode.mydomain.com:8020", displayPriority = 2) void setFileSystemUri(final String hdfsUri); @PipelineProperty(description = "The user to connect to HDFS as", displayPriority = 3) void setRunAsUser(final String runAsUser); @SuppressWarnings("unused") @PipelineProperty(description = "When the current output file exceeds this size it will be closed and a new one created.", displayPriority = 4) void setRollSize(final String size); @PipelineProperty(description = "Choose if you want to split aggregated streams into separate output files.", defaultValue = "false", displayPriority = 5) void setSplitAggregatedStreams(final boolean splitAggregatedStreams); @PipelineProperty(description = "Choose if you want to split individual records into separate output files.", defaultValue = "false", displayPriority = 6) void setSplitRecords(final boolean splitRecords);  }
@Disabled @Test void basicTest() throws IOException { final Path path = new Path("/dateTest.txt"); HDFSFileAppender.runOnHDFS(userGroupInformation, conf, hdfs -> { try { hdfs.exists(path); } catch (final IOException e) { throw new UncheckedIOException(e); } }); final ContentSummary summary = hdfs.getContentSummary(path); System.out.println(summary); }