Update with workerFileId.

core/server/src/main/java/alluxio/worker/file/FileSystemWorker.java

@@ -49,7 +49,7 @@ public final class FileSystemWorker extends AbstractWorker {
   private final Configuration mConf;
   /** Logic for handling RPC requests. */
   private final FileSystemWorkerClientServiceHandler mServiceHandler;
-  /** Manager for under file system operations */
+  /** Manager for under file system operations. */
   private final UnderFileSystemManager mUnderFileSystemManager;
 
   /** The service that persists files. */
@@ -115,14 +115,41 @@ public void stop() {
     getExecutorService().shutdown();
   }
 
-  public void ufsCancelFile(String path) throws FileDoesNotExistException, IOException {
-    mUnderFileSystemManager.cancelFile(path);
+  /**
+   * Cancels a file currently being written to the under file system. The open stream will be
+   * closed and the partial file will be cleaned up.
+   *
+   * @param workerFileId the id of the file to cancel, only understood by the worker that created
+   *                     the file
+   * @throws FileDoesNotExistException if this worker is not writing the specified file
+   * @throws IOException if an error occurs interacting with the under file system
+   */
+  public void ufsCancelFile(long workerFileId) throws FileDoesNotExistException, IOException {
+    mUnderFileSystemManager.cancelFile(workerFileId);
   }
 
-  public void ufsCompleteFile(String path) throws FileDoesNotExistException, IOException {
-    mUnderFileSystemManager.completeFile(path);
+  /**
+   * Completes a file currently being written to the under file system. The open stream will be
+   * closed and the partial file will be promoted to the completed file in the under file system.
+   *
+   * @param workerFileId the id of the file to cancel, only understood by the worker that created
+   *                     the file
+   * @throws FileDoesNotExistException if the worker is not writing the specified file
+   * @throws IOException if an error occurs interacting with the under file system
+   */
+  public void ufsCompleteFile(long workerFileId) throws FileDoesNotExistException, IOException {
+    mUnderFileSystemManager.completeFile(workerFileId);
   }
 
+  /**
+   * Creates a new file in the under file system. This will register a new stream in the under
+   * file system manager. The stream can only be accessed with the returned id afterward.
+   *
+   * @param path the path of the file to create
+   * @throws FileAlreadyExistsException if a file already exists in the under file system with
+   *                                    the same path
+   * @throws IOException if an error occurs interacting with the under file system
+   */
   public void ufsCreateFile(String path) throws FileAlreadyExistsException, IOException {
     mUnderFileSystemManager.createFile(path);
   }

core/server/src/main/java/alluxio/worker/file/FileSystemWorkerClientServiceHandler.java

@@ -47,10 +47,10 @@ public long getServiceVersion() {
   }
 
   @Override
-  public void ufsCancelFile(String path, UFSCancelFileTOptions options)
+  public void ufsCancelFile(long workerFileId, UFSCancelFileTOptions options)
       throws AlluxioTException, ThriftIOException {
     try {
-      mWorker.ufsCancelFile(path);
+      mWorker.ufsCancelFile(workerFileId);
     } catch (IOException e) {
       throw new ThriftIOException(e.getMessage());
     } catch (AlluxioException e) {
@@ -59,10 +59,10 @@ public void ufsCancelFile(String path, UFSCancelFileTOptions options)
   }
 
   @Override
-  public void ufsCompleteFile(String path, UFSCompleteFileTOptions options)
+  public void ufsCompleteFile(long workerFileId, UFSCompleteFileTOptions options)
       throws AlluxioTException, ThriftIOException {
     try {
-      mWorker.ufsCompleteFile(path);
+      mWorker.ufsCompleteFile(workerFileId);
     } catch (IOException e) {
       throw new ThriftIOException(e.getMessage());
     } catch (AlluxioException e) {

core/server/src/main/java/alluxio/worker/file/UnderFileSystemManager.java

@@ -17,38 +17,54 @@
  * file names to open streams.
  */
 public class UnderFileSystemManager {
-  private ConcurrentMap<String, OutputStream> mFileToOutputStreamMap;
+  private class NamedOutputStream {
+    private final OutputStream mStream;
+    private final String mPath;
+
+    private NamedOutputStream(OutputStream stream, String path) {
+      mStream = stream;
+      mPath = path;
+    }
+
+    public OutputStream getStream() {
+      return mStream;
+    }
+
+    public String getPath() {
+      return mPath;
+    }
+  }
+
+
+  private ConcurrentMap<Long, NamedOutputStream> mStreams;
 
   public UnderFileSystemManager() {
-    mFileToOutputStreamMap = new ConcurrentHashMap<>();
+    mStreams = new ConcurrentHashMap<>();
   }
 
   public OutputStream createFile(String path) throws FileAlreadyExistsException, IOException {
-    OutputStream stream = mFileToOutputStreamMap.get(path);
-    if (stream == null) {
-      UnderFileSystem ufs = UnderFileSystem.get(path, WorkerContext.getConf());
-      OutputStream newStream = ufs.create(path);
-      stream = mFileToOutputStreamMap.putIfAbsent(path, newStream);
-      if (stream == null) {
-        return newStream;
-      }
+    UnderFileSystem ufs = UnderFileSystem.get(path, WorkerContext.getConf());
+    if (ufs.exists(path)) {
+      throw new FileAlreadyExistsException(ExceptionMessage.FAILED_UFS_CREATE.getMessage(path));
     }
-    throw new FileAlreadyExistsException(ExceptionMessage.FAILED_UFS_CREATE.getMessage(path));
+
+    OutputStream newStream = ufs.create(path);
+    stream = mStreams.putIfAbsent(path, newStream);
   }
 
-  public void cancelFile(String path) throws FileDoesNotExistException, IOException {
+  public void cancelFile(long workerFileId) throws FileDoesNotExistException, IOException {
     closeFile(path);
     UnderFileSystem ufs = UnderFileSystem.get(path, WorkerContext.getConf());
     ufs.delete(path, false);
   }
 
-  public void completeFile(String path) throws FileDoesNotExistException, IOException {
+  public void completeFile(long workerFileId) throws FileDoesNotExistException, IOException {
     closeFile(path);
   }
 
   // TODO(calvin): Make the exception accurate.
-  private void closeFile(String path) throws FileDoesNotExistException, IOException {
-    OutputStream stream = mFileToOutputStreamMap.remove(path);
+  private void closeFile(long workerFileId) throws FileDoesNotExistException, IOException {
+    OutputStream stream = mStreams.remove(workerFileId);
     if (stream != null) {
       stream.close();
     } else {