@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);
    FileStatus status;
    try {
      status = getFileStatus(client, file);
    } catch (FileNotFoundException fnfe) {
      status = null;
    }
    if (status != null) {
      if (overwrite && !status.isDirectory()) {
        delete(client, file, false);
      } else {
        disconnect(client);
        throw new FileAlreadyExistsException("File already exists: " + file);
      }
    }
    
    Path parent = absolute.getParent();
    if (parent == null || !mkdirs(client, parent, FsPermission.getDirDefault())) {
      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());
    OutputStream outputStream = client.storeFileStream(file.getName());

    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
      if (outputStream != null) {
        IOUtils.closeStream(outputStream);
      }
      disconnect(client);
      throw new IOException("Unable to create file: " + file + ", Aborting");
    }

    FSDataOutputStream fos = new FSDataOutputStream(outputStream, 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());
        }
      }
    };
    return fos;
  }
@Test
  public void testCreateWithoutWritePermissions() throws Exception {
    BaseUser user = server.addUser("test", "password");
    Configuration configuration = new Configuration();
    configuration.set("fs.defaultFS", "ftp:///");
    configuration.set("fs.ftp.host", "localhost");
    configuration.setInt("fs.ftp.host.port", server.getPort());
    configuration.set("fs.ftp.user.localhost", user.getName());
    configuration.set("fs.ftp.password.localhost", user.getPassword());
    configuration.setBoolean("fs.ftp.impl.disable.cache", true);

    FileSystem fs = FileSystem.get(configuration);
    byte[] bytesExpected = "hello world".getBytes(StandardCharsets.UTF_8);
    LambdaTestUtils.intercept(
        IOException.class, "Unable to create file: test1.txt, Aborting",
        () -> {
          try (FSDataOutputStream out = fs.create(new Path("test1.txt"))) {
            out.write(bytesExpected);
          }
        }
    );
  }