private HDFSLockedOutputStream getHDFSLockedOutputStream(final Path filePath) throws IOException {
        final UserGroupInformation ugi = getUserGroupInformation();
        HDFSLockedOutputStream hdfsLockedOutputStream = null;

        try {
            hdfsLockedOutputStream = ugi.doAs(new PrivilegedExceptionAction<HDFSLockedOutputStream>() {
                @Override
                public HDFSLockedOutputStream run() {
                    try {
                        final FileSystem hdfs = getHDFS();
                        final Path dir = filePath.getParent();

                        // Create the directory if it doesn't exist
                        if (!hdfs.exists(dir)) {
                            hdfs.mkdirs(dir);
                        }

                        final Path lockFile = createCleanPath(filePath + LOCK_EXTENSION);
                        final Path outFile = filePath;

                        // Make sure we can create both output files without
                        // overwriting
                        // another
                        // file.
                        if (hdfs.exists(lockFile)) {
                            throw ProcessException.create("Output file \"" + lockFile.toString() + "\" already exists");
                        }

                        if (hdfs.exists(outFile)) {
                            throw ProcessException.create("Output file \"" + outFile.toString() + "\" already exists");
                        }

                        // Get a writer for the new lock file.
                        final OutputStream outputStream = new BufferedOutputStream(hdfs.create(lockFile));

                        return new HDFSLockedOutputStream(outputStream, lockFile, outFile, hdfs);
                    } catch (final IOException e) {
                        throw new UncheckedIOException(e);
                    }
                }
            });
        } catch (final InterruptedException e) {
            // Continue to interrupt this thread.
            Thread.currentThread().interrupt();
        }

        if (hdfsLockedOutputStream == null) {
            throw new RuntimeException(String.format(
                    "Something went wrong creating the HDFSLockedOutputStream, outFile %s, hdfs uri %s",
                    filePath, hdfsUri));
        }

        return hdfsLockedOutputStream;
    }
HDFS does not support some chars in its dir/filenames so need to clean
them here

@param path The path to clean
@return A cleaned path
stroom-pipeline/src/main/java/stroom/pipeline/writer/HDFSFileAppender.java