code: @Override
    public FSDataOutputStream create(Path file, FsPermission permission,
        boolean overwrite, int bufferSize, short replication, long blockSize,
        Progressable progress) throws IOException {
      final FTPClient client = connect();
      Path workDir = new Path(client.printWorkingDirectory());
      Path absolute = makeAbsolute(workDir, file);
      if (exists(client, file)) {
        if (overwrite) {
          delete(client, file);
        } else {
          disconnect(client);
          throw new IOException("File already exists: " + file);
        }
      }
      Path parent = absolute.getParent();
      if (parent == null || !mkdirs(client, parent, FsPermission.getDefault())) {
        parent = (parent == null) ? new Path("/") : parent;
        disconnect(client);
        throw new IOException("create(): Mkdirs failed to create: " + parent);
      }
      client.allocate(bufferSize);
      // Change to parent directory on the server. Only then can we write to the
      // file on the server by opening up an OutputStream. As a side effect the
      // working directory on the server is changed to the parent directory of the
      // file. The FTP client connection is closed when close() is called on the
      // FSDataOutputStream.
      client.changeWorkingDirectory(parent.toUri().getPath());
      FSDataOutputStream fos = new FSDataOutputStream(client.storeFileStream(file
          .getName()), statistics) {
        @Override
        public void close() throws IOException {
          super.close();
          if (!client.isConnected()) {
            throw new FTPException("Client not connected");
          }
          boolean cmdCompleted = client.completePendingCommand();
          disconnect(client);
          if (!cmdCompleted) {
            throw new FTPException("Could not complete transfer, Reply Code - "
                + client.getReplyCode());
          }
        }
      };
      if (!FTPReply.isPositivePreliminary(client.getReplyCode())) {
        // The ftpClient is an inconsistent state. Must close the stream
        // which in turn will logout and disconnect from FTP server
        fos.close();
        throw new IOException("Unable to create file: " + file + ", Aborting");
      }
      return fos;
    }
docs: A stream obtained via this call must be closed before using other APIs of
  this class or else the invocation will block.