FTPFileSystem extends FileSystem { @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); client.changeWorkingDirectory(parent.toUri().getPath()); OutputStream outputStream = client.storeFileStream(file.getName()); if (!FTPReply.isPositivePreliminary(client.getReplyCode())) { 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; }  @Override String getScheme(); @Override void initialize(URI uri, Configuration conf); @Override FSDataInputStream open(Path file, int bufferSize); @Override FSDataOutputStream create(Path file, FsPermission permission,
      boolean overwrite, int bufferSize, short replication, long blockSize,
      Progressable progress); @Override FSDataOutputStream append(Path f, int bufferSize,
      Progressable progress); @Override boolean delete(Path file, boolean recursive); @Override URI getUri(); @Override FileStatus[] listStatus(Path file); @Override FileStatus getFileStatus(Path file); @Override boolean mkdirs(Path file, FsPermission permission); @Override boolean rename(Path src, Path dst); @Override Path getWorkingDirectory(); @Override Path getHomeDirectory(); @Override void setWorkingDirectory(Path newDir); static final Logger LOG; static final int DEFAULT_BUFFER_SIZE; static final int DEFAULT_BLOCK_SIZE; static final long DEFAULT_TIMEOUT; static final String FS_FTP_USER_PREFIX; static final String FS_FTP_HOST; static final String FS_FTP_HOST_PORT; static final String FS_FTP_PASSWORD_PREFIX; static final String FS_FTP_DATA_CONNECTION_MODE; static final String FS_FTP_TRANSFER_MODE; static final String E_SAME_DIRECTORY_ONLY; static final String FS_FTP_TIMEOUT; }
@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); } } ); }