diff --git a/src/main/java/org/apache/commons/net/ftp/FTP.java b/src/main/java/org/apache/commons/net/ftp/FTP.java
index 0c0b3044d..a956e6a15 100644
--- a/src/main/java/org/apache/commons/net/ftp/FTP.java
+++ b/src/main/java/org/apache/commons/net/ftp/FTP.java
@@ -151,6 +151,13 @@ public class FTP extends SocketClient {
      */
     public static final int COMPRESSED_TRANSFER_MODE = 12;
 
+    /**
+     * A constant used to indicate a file is to be transferred as FTP compressed
+     * data with MODE Z (zlib). All constants ending in <code>TRANSFER_MODE</code> are used to indicate
+     * file transfer modes.
+     */
+    public static final int COMPRESSED_MODE_Z_TRANSFER_MODE = 13;
+
     // We have to ensure that the protocol communication is in ASCII,
     // but we use ISO-8859-1 just in case 8-bit characters cross
     // the wire.
@@ -163,7 +170,7 @@ public class FTP extends SocketClient {
     /** Length of the FTP reply code (3 alphanumerics) */
     public static final int REPLY_CODE_LEN = 3;
 
-    private static final String modes = "AEILNTCFRPSBC";
+    private static final String modes = "AEILNTCFRPSBCZ";
     protected int _replyCode;
     protected ArrayList<String> _replyLines;
     protected boolean _newReplyString;
diff --git a/src/main/java/org/apache/commons/net/ftp/FTPClient.java b/src/main/java/org/apache/commons/net/ftp/FTPClient.java
index f808f6f8d..5babcb6fb 100644
--- a/src/main/java/org/apache/commons/net/ftp/FTPClient.java
+++ b/src/main/java/org/apache/commons/net/ftp/FTPClient.java
@@ -507,7 +507,6 @@ static String parsePathname(final String reply) {
     private int fileFormat;
     @SuppressWarnings("unused") // field is written, but currently not read
     private int fileStructure;
-    @SuppressWarnings("unused") // field is written, but currently not read
     private int fileTransferMode;
 
     private boolean remoteVerificationEnabled;
@@ -713,7 +712,8 @@ protected Socket _openDataConnection_(final String command, final String arg) th
                 if (soTimeoutMillis >= 0) {
                     server.setSoTimeout(soTimeoutMillis);
                 }
-                socket = server.accept();
+
+                socket = wrapSocketIfModeZisEnabled(server.accept());
 
                 // Ensure the timeout is set before any commands are issued on the new socket
                 if (soTimeoutMillis >= 0) {
@@ -749,7 +749,8 @@ protected Socket _openDataConnection_(final String command, final String arg) th
                 _parsePassiveModeReply(_replyLines.get(0));
             }
 
-            socket = _socketFactory_.createSocket();
+            socket = wrapSocketIfModeZisEnabled(_socketFactory_.createSocket());
+
             if (receiveDataSocketBufferSize > 0) {
                 socket.setReceiveBufferSize(receiveDataSocketBufferSize);
             }
@@ -793,6 +794,14 @@ protected Socket _openDataConnection_(final String command, final String arg) th
         return socket;
     }
 
+    private Socket wrapSocketIfModeZisEnabled(final Socket plainSocket) {
+        if (fileTransferMode == COMPRESSED_MODE_Z_TRANSFER_MODE) {
+            return ModeZSocket.wrap(plainSocket);
+        } else {
+            return plainSocket;
+        }
+    }
+
     protected void _parseExtendedPassiveModeReply(String reply) throws MalformedServerReplyException {
         reply = reply.substring(reply.indexOf('(') + 1, reply.indexOf(')')).trim();
 
diff --git a/src/main/java/org/apache/commons/net/ftp/ModeZSocket.java b/src/main/java/org/apache/commons/net/ftp/ModeZSocket.java
new file mode 100644
index 000000000..2d79b634d
--- /dev/null
+++ b/src/main/java/org/apache/commons/net/ftp/ModeZSocket.java
@@ -0,0 +1,252 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.commons.net.ftp;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.nio.channels.SocketChannel;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.InflaterInputStream;
+
+/**
+ * Wrapper class for FTP data channel sockets when using MODE Z. All methods
+ * except of {@link #getInputStream()} and {@link #getOutputStream()} are
+ * calling the delegate methods directly.
+ */
+public class ModeZSocket extends Socket {
+
+    private final Socket delegate;
+
+    private ModeZSocket(final Socket delegate) {
+        this.delegate = delegate;
+    }
+
+    static Socket wrap(final Socket plain) {
+        return new ModeZSocket(plain);
+    }
+
+    @Override
+    public void connect(final SocketAddress endpoint) throws IOException {
+        delegate.connect(endpoint);
+    }
+
+    @Override
+    public void connect(final SocketAddress endpoint, final int timeout) throws IOException {
+        delegate.connect(endpoint, timeout);
+    }
+
+    @Override
+    public void bind(final SocketAddress bindpoint) throws IOException {
+        delegate.bind(bindpoint);
+    }
+
+    @Override
+    public InetAddress getInetAddress() {
+        return delegate.getInetAddress();
+    }
+
+    @Override
+    public InetAddress getLocalAddress() {
+        return delegate.getLocalAddress();
+    }
+
+    @Override
+    public int getPort() {
+        return delegate.getPort();
+    }
+
+    @Override
+    public int getLocalPort() {
+        return delegate.getLocalPort();
+    }
+
+    @Override
+    public SocketAddress getRemoteSocketAddress() {
+        return delegate.getRemoteSocketAddress();
+    }
+
+    @Override
+    public SocketAddress getLocalSocketAddress() {
+        return delegate.getLocalSocketAddress();
+    }
+
+    @Override
+    public SocketChannel getChannel() {
+        return delegate.getChannel();
+    }
+
+    @Override
+    public InputStream getInputStream() throws IOException {
+        return new InflaterInputStream(delegate.getInputStream());
+    }
+
+    @Override
+    public OutputStream getOutputStream() throws IOException {
+        return new DeflaterOutputStream(delegate.getOutputStream());
+    }
+
+    @Override
+    public void setTcpNoDelay(final boolean on) throws SocketException {
+        delegate.setTcpNoDelay(on);
+    }
+
+    @Override
+    public boolean getTcpNoDelay() throws SocketException {
+        return delegate.getTcpNoDelay();
+    }
+
+    @Override
+    public void setSoLinger(final boolean on, final int linger) throws SocketException {
+        delegate.setSoLinger(on, linger);
+    }
+
+    @Override
+    public int getSoLinger() throws SocketException {
+        return delegate.getSoLinger();
+    }
+
+    @Override
+    public void sendUrgentData(final int data) throws IOException {
+        delegate.sendUrgentData(data);
+    }
+
+    @Override
+    public void setOOBInline(final boolean on) throws SocketException {
+        delegate.setOOBInline(on);
+    }
+
+    @Override
+    public boolean getOOBInline() throws SocketException {
+        return delegate.getOOBInline();
+    }
+
+    @Override
+    public synchronized void setSoTimeout(final int timeout) throws SocketException {
+        delegate.setSoTimeout(timeout);
+    }
+
+    @Override
+    public synchronized int getSoTimeout() throws SocketException {
+        return delegate.getSoTimeout();
+    }
+
+    @Override
+    public synchronized void setSendBufferSize(final int size) throws SocketException {
+        delegate.setSendBufferSize(size);
+    }
+
+    @Override
+    public synchronized int getSendBufferSize() throws SocketException {
+        return delegate.getSendBufferSize();
+    }
+
+    @Override
+    public synchronized void setReceiveBufferSize(final int size) throws SocketException {
+        delegate.setReceiveBufferSize(size);
+    }
+
+    @Override
+    public synchronized int getReceiveBufferSize() throws SocketException {
+        return delegate.getReceiveBufferSize();
+    }
+
+    @Override
+    public void setKeepAlive(final boolean on) throws SocketException {
+        delegate.setKeepAlive(on);
+    }
+
+    @Override
+    public boolean getKeepAlive() throws SocketException {
+        return delegate.getKeepAlive();
+    }
+
+    @Override
+    public void setTrafficClass(final int tc) throws SocketException {
+        delegate.setTrafficClass(tc);
+    }
+
+    @Override
+    public int getTrafficClass() throws SocketException {
+        return delegate.getTrafficClass();
+    }
+
+    @Override
+    public void setReuseAddress(final boolean on) throws SocketException {
+        delegate.setReuseAddress(on);
+    }
+
+    @Override
+    public boolean getReuseAddress() throws SocketException {
+        return delegate.getReuseAddress();
+    }
+
+    @Override
+    public synchronized void close() throws IOException {
+        delegate.close();
+    }
+
+    @Override
+    public void shutdownInput() throws IOException {
+        delegate.shutdownInput();
+    }
+
+    @Override
+    public void shutdownOutput() throws IOException {
+        delegate.shutdownOutput();
+    }
+
+    @Override
+    public String toString() {
+        return delegate.toString();
+    }
+
+    @Override
+    public boolean isConnected() {
+        return delegate.isConnected();
+    }
+
+    @Override
+    public boolean isBound() {
+        return delegate.isBound();
+    }
+
+    @Override
+    public boolean isClosed() {
+        return delegate.isClosed();
+    }
+
+    @Override
+    public boolean isInputShutdown() {
+        return delegate.isInputShutdown();
+    }
+
+    @Override
+    public boolean isOutputShutdown() {
+        return delegate.isOutputShutdown();
+    }
+
+    @Override
+    public void setPerformancePreferences(final int connectionTime, final int latency, final int bandwidth) {
+        delegate.setPerformancePreferences(connectionTime, latency, bandwidth);
+    }
+}
diff --git a/src/test/java/org/apache/commons/net/ftp/FTPClientModeZTest.java b/src/test/java/org/apache/commons/net/ftp/FTPClientModeZTest.java
new file mode 100644
index 000000000..a47d12495
--- /dev/null
+++ b/src/test/java/org/apache/commons/net/ftp/FTPClientModeZTest.java
@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.commons.net.ftp;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.nio.file.Files.readAllBytes;
+import static java.nio.file.Files.write;
+import static org.apache.commons.io.FileUtils.deleteDirectory;
+import static org.apache.commons.net.ftp.FTP.COMPRESSED_MODE_Z_TRANSFER_MODE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.ftpserver.FtpServer;
+import org.apache.ftpserver.FtpServerFactory;
+import org.apache.ftpserver.ftplet.Authority;
+import org.apache.ftpserver.ftplet.FtpException;
+import org.apache.ftpserver.ftplet.UserManager;
+import org.apache.ftpserver.listener.Listener;
+import org.apache.ftpserver.listener.ListenerFactory;
+import org.apache.ftpserver.usermanager.PropertiesUserManagerFactory;
+import org.apache.ftpserver.usermanager.impl.BaseUser;
+import org.apache.ftpserver.usermanager.impl.WritePermission;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+@RunWith(Parameterized.class)
+public class FTPClientModeZTest {
+
+    private static final String DEFAULT_HOME = "ftp_root/";
+    private final boolean useLocalPassiveFTP;
+
+    @Parameters(name = "useLocalPassiveFTP={0}")
+    public static Boolean[] testParameters() {
+        return new Boolean[] { true, false };
+    }
+
+    public FTPClientModeZTest(boolean useLocalPassiveFTP) {
+        this.useLocalPassiveFTP = useLocalPassiveFTP;
+    }
+
+    @Test
+    public void testRetrievingFiles() throws Exception {
+
+        new File(DEFAULT_HOME).mkdirs();
+        String filename = "test_download.txt";
+        String fileContent = "Created at " + Instant.now();
+        write(Paths.get(DEFAULT_HOME).resolve(filename), fileContent.getBytes(UTF_8));
+
+        runWithFTPserver((port, user, password) -> {
+            FTPClient client = new FTPClient();
+            try {
+                client.connect("localhost", port);
+                client.login(user, password);
+                if (useLocalPassiveFTP) {
+                    client.enterLocalPassiveMode();
+                }
+                assertTrue("Mode Z successfully activated",
+                        client.setFileTransferMode(COMPRESSED_MODE_Z_TRANSFER_MODE));
+
+                FTPFile[] files = client.listFiles();
+                assertEquals("Only single file in home directory", 1, files.length);
+
+                ByteArrayOutputStream bos = new ByteArrayOutputStream();
+                assertTrue("File successfully transferred", client.retrieveFile(files[0].getName(), bos));
+                assertEquals("File content is not corrupted", fileContent, new String(bos.toByteArray(), UTF_8));
+            } finally {
+                client.logout();
+            }
+        });
+    }
+
+    @Test
+    public void testStoringFiles() throws Exception {
+
+        runWithFTPserver((port, user, password) -> {
+            FTPClient client = new FTPClient();
+            try {
+                client.connect("localhost", port);
+                client.login(user, password);
+                if (useLocalPassiveFTP) {
+                    client.enterLocalPassiveMode();
+                }
+                assertTrue("Mode Z successfully activated",
+                        client.setFileTransferMode(COMPRESSED_MODE_Z_TRANSFER_MODE));
+
+                FTPFile[] filesBeforeUpload = client.listFiles();
+                assertEquals("No files in home directory before upload", 0, filesBeforeUpload.length);
+
+                String filename = "test_upload.txt";
+                String fileContent = "Created at " + Instant.now();
+                assertTrue("File successfully transferred",
+                        client.storeFile(filename, new ByteArrayInputStream(fileContent.getBytes(UTF_8))));
+
+                FTPFile[] filesAfterUpload = client.listFiles();
+                assertEquals("Single file in home directory after upload", 1, filesAfterUpload.length);
+
+                Path p = Paths.get(DEFAULT_HOME, filename);
+                assertEquals("File content is not corrupted", fileContent, new String(readAllBytes(p), UTF_8));
+            } finally {
+                client.logout();
+            }
+        });
+    }
+
+    @Before
+    @After
+    public void cleanup() throws IOException {
+        deleteDirectory(new File(DEFAULT_HOME));
+    }
+
+    private static final class FtpServerAndPort {
+        private final int port;
+        private final FtpServer ftpServer;
+
+        FtpServerAndPort(FtpServer ftpServer, int port) {
+            this.port = port;
+            this.ftpServer = ftpServer;
+        }
+    }
+
+    @FunctionalInterface
+    interface Runner {
+        void run(int port, String user, String password) throws Exception;
+    }
+
+    private static void runWithFTPserver(Runner runner) throws Exception {
+
+        String username = "test";
+        String password = "test";
+        FtpServerAndPort ftpServerAndPort = setupPlainFTPserver(username, password);
+        try {
+            runner.run(ftpServerAndPort.port, username, password);
+        } finally {
+            ftpServerAndPort.ftpServer.stop();
+        }
+    }
+
+    private static FtpServerAndPort setupPlainFTPserver(final String username, final String password)
+            throws FtpException {
+
+        final FtpServerFactory serverFactory = new FtpServerFactory();
+
+        // Init user
+        serverFactory.setUserManager(initUserManager(username, password));
+
+        final ListenerFactory factory = new ListenerFactory();
+        // Automatically assign port.
+        factory.setPort(0);
+
+        // replace the default listener
+        Listener listener = factory.createListener();
+        serverFactory.addListener("default", listener);
+
+        // start the server
+        FtpServer server = serverFactory.createServer();
+        server.start();
+
+        return new FtpServerAndPort(server, listener.getPort());
+    }
+
+    private static UserManager initUserManager(final String username, final String password) throws FtpException {
+
+        final PropertiesUserManagerFactory propertiesUserManagerFactory = new PropertiesUserManagerFactory();
+        final UserManager userManager = propertiesUserManagerFactory.createUserManager();
+        BaseUser user = new BaseUser();
+        user.setName(username);
+        user.setPassword(password);
+
+        List<Authority> authorities = new ArrayList<Authority>();
+        authorities.add(new WritePermission());
+        user.setAuthorities(authorities);
+
+        new File(DEFAULT_HOME).mkdirs();
+        user.setHomeDirectory(DEFAULT_HOME);
+        userManager.save(user);
+        return userManager;
+    }
+
+}