INT-3593: (S)FTP OG Partial Updates (mget/mput)

JIRA: https://jira.spring.io/browse/INT-3593

Throw a `PartialSuccessException` if an exception occurs after partial success (some files transferred).

spring-integration-core/src/main/java/org/springframework/integration/support/PartialSuccessException.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed 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.springframework.integration.support;
+
+import java.util.Collection;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
+
+import reactor.core.support.Assert;
+
+/**
+ * A {@link MessagingException} thrown when a non-transactional operation is
+ * performing multiple updates from a single message, e.g. an FTP 'mput' operation.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class PartialSuccessException extends MessagingException {
+
+	private static final long serialVersionUID = 8810900575763284993L;
+
+	private final Collection<?> partialResults;
+
+	private final Collection<?> derivedInput;
+
+	/**
+	 *
+	 * @param message the message.
+	 * @param description the description.
+	 * @param cause the cause.
+	 * @param partialResults The subset of multiple updates that were successful before the cause occurred.
+	 * @param derivedInput The collection (usually derived from the message) of input data; e.g. a filtered
+	 * list of local files being sent to FTP using {@code mput}.
+	 */
+	public PartialSuccessException(Message<?> message, String description, Throwable cause,
+			Collection<?> partialResults, Collection<?> derivedInput) {
+		super(message, description, cause);
+		Assert.notNull(cause, "Cause is required");
+		this.partialResults = partialResults;
+		this.derivedInput = derivedInput;
+	}
+
+	/**
+	 * See {@link #PartialSuccessException(Message, String, Throwable, Collection, Collection)}.
+	 * @return the partial results
+	 */
+	public Collection<?> getPartialResults() {
+		return this.partialResults;
+	}
+
+	/**
+	 * See {@link #PartialSuccessException(Message, String, Throwable, Collection, Collection)}.
+	 * @return the derived input.
+	 */
+	public Collection<?> getDerivedInput() {
+		return this.derivedInput;
+	}
+
+	/**
+	 * Convenience version of {@link #getPartialResults()} to avoid casting
+	 * @param clazz the type.
+	 * @param <T> the result type.
+	 * @return the partial results.
+	 */
+	@SuppressWarnings("unchecked")
+	public <T> Collection<T> getPartialResults(Class<T> clazz) {
+		return (Collection<T>) this.partialResults;
+	}
+
+	/**
+	 * Convenience version of {@link #getDerivedInput()} to avoid casting
+	 * @param clazz the type.
+	 * @param <T> the type of input.
+	 * @return the partial results.
+	 */
+	@SuppressWarnings("unchecked")
+	public <T> Collection<T> getDerivedInput(Class<T> clazz) {
+		return (Collection<T>) this.derivedInput;
+	}
+
+	@Override
+	public String toString() {
+		return "PartialSuccessException [" + getMessage() + ":" + getCause().getMessage()
+				+ ", partialResults=" + partialResults + ", derivedInput=" + derivedInput
+				+ ", failedMessage=" + getFailedMessage() + "]";
+	}
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java

@@ -40,7 +40,6 @@ import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
-import org.springframework.integration.file.remote.RemoteFileUtils;
 import org.springframework.integration.file.remote.SessionCallback;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
@@ -48,6 +47,7 @@ import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.integration.support.PartialSuccessException;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
@@ -568,25 +568,42 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		File[] files = file.listFiles();
 		List<File> filteredFiles = this.filterMputFiles(files);
 		List<String> replies = new ArrayList<String>();
-		for (File filteredFile : filteredFiles) {
-			if (!filteredFile.isDirectory()) {
-				String path = this.doPut(this.getMessageBuilderFactory().withPayload(filteredFile)
-						.copyHeaders(requestMessage.getHeaders())
-						.build(), subDirectory);
-				if (path == null) {//NOSONAR - false positive
-					if (logger.isDebugEnabled()) {
-						logger.debug("File " + filteredFile.getAbsolutePath() + " removed before transfer; ignoring");
+		try {
+			for (File filteredFile : filteredFiles) {
+				if (!filteredFile.isDirectory()) {
+					String path = this.doPut(this.getMessageBuilderFactory().withPayload(filteredFile)
+							.copyHeaders(requestMessage.getHeaders())
+							.build(), subDirectory);
+					if (path == null) {//NOSONAR - false positive
+						if (logger.isDebugEnabled()) {
+							logger.debug("File " + filteredFile.getAbsolutePath() + " removed before transfer; ignoring");
+						}
+					}
+					else {
+						replies.add(path);
 					}
 				}
-				else {
-					replies.add(path);
+				else if (this.options.contains(Option.RECURSIVE)){
+					String newSubDirectory = (StringUtils.hasText(subDirectory) ?
+							subDirectory + this.remoteFileTemplate.getRemoteFileSeparator() : "")
+						+ filteredFile.getName();
+					replies.addAll(this.putLocalDirectory(requestMessage, filteredFile, newSubDirectory));
 				}
 			}
-			else if (this.options.contains(Option.RECURSIVE)){
-				String newSubDirectory = (StringUtils.hasText(subDirectory) ?
-						subDirectory + this.remoteFileTemplate.getRemoteFileSeparator() : "")
-					+ filteredFile.getName();
-				replies.addAll(this.putLocalDirectory(requestMessage, filteredFile, newSubDirectory));
+		}
+		catch (Exception e) {
+			if (replies.size() > 0) {
+				throw new PartialSuccessException(requestMessage,
+						"Partially successful 'mput' operation" + (subDirectory == null ? "" : (" on " + subDirectory)),
+						e, replies, filteredFiles);
+			}
+			else if (e instanceof PartialSuccessException) {
+				throw new PartialSuccessException(requestMessage,
+						"Partially successful 'mput' operation" + (subDirectory == null ? "" : (" on " + subDirectory)),
+						e, replies, filteredFiles);
+			}
+			else if (e instanceof MessagingException) {
+				throw (MessagingException) e;
 			}
 		}
 		return replies;
@@ -792,18 +809,33 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		}
 		List<File> files = new ArrayList<File>();
 		String remoteFileSeparator = this.remoteFileTemplate.getRemoteFileSeparator();
-		for (String fileName : fileNames) {
-			File file;
-			if (fileName.contains(remoteFileSeparator) &&
-					fileName.startsWith(remoteDirectory)) { // the server returned the full path
-				file = this.get(message, session, remoteDirectory, fileName,
-						fileName.substring(fileName.lastIndexOf(remoteFileSeparator)), false);
+		try {
+			for (String fileName : fileNames) {
+				File file;
+				if (fileName.contains(remoteFileSeparator) &&
+						fileName.startsWith(remoteDirectory)) { // the server returned the full path
+					file = this.get(message, session, remoteDirectory, fileName,
+							fileName.substring(fileName.lastIndexOf(remoteFileSeparator)), false);
+				}
+				else {
+					file = this.get(message, session, remoteDirectory,
+							this.generateFullPath(remoteDirectory, fileName), fileName, false);
+				}
+				files.add(file);
 			}
-			else {
-				file = this.get(message, session, remoteDirectory,
-						this.generateFullPath(remoteDirectory, fileName), fileName, false);
+		}
+		catch (Exception e) {
+			if (files.size() > 0) {
+				throw new PartialSuccessException(message,
+						"Partially successful 'mget' operation on " + remoteDirectory, e, files,
+						Arrays.asList(fileNames));
+			}
+			else if (e instanceof MessagingException) {
+				throw (MessagingException) e;
+			}
+			else if (e instanceof IOException) {
+				throw (IOException) e;
 			}
-			files.add(file);
 		}
 		return files;
 	}
@@ -817,17 +849,31 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 			throw new MessagingException("No files found at " + remoteDirectory
 					+ " with pattern " + remoteFilename);
 		}
-		for (AbstractFileInfo<F> lsEntry : fileNames) {
-			String fullFileName = remoteDirectory + this.getFilename(lsEntry);
-			/*
-			 * With recursion, the filename might contain subdirectory information
-			 * normalize each file separately.
-			 */
-			String fileName = this.getRemoteFilename(fullFileName);
-			String actualRemoteDirectory = this.getRemoteDirectory(fullFileName, fileName);
-			File file = this.get(message, session, actualRemoteDirectory,
-						fullFileName, fileName, false);
-			files.add(file);
+		try {
+			for (AbstractFileInfo<F> lsEntry : fileNames) {
+				String fullFileName = remoteDirectory + this.getFilename(lsEntry);
+				/*
+				 * With recursion, the filename might contain subdirectory information
+				 * normalize each file separately.
+				 */
+				String fileName = this.getRemoteFilename(fullFileName);
+				String actualRemoteDirectory = this.getRemoteDirectory(fullFileName, fileName);
+				File file = this.get(message, session, actualRemoteDirectory,
+							fullFileName, fileName, false);
+				files.add(file);
+			}
+		}
+		catch (Exception e) {
+			if (files.size() > 0) {
+				throw new PartialSuccessException(message,
+						"Partially successful recursive 'mget' operation on " + remoteDirectory, e, files, fileNames);
+			}
+			else if (e instanceof MessagingException) {
+				throw (MessagingException) e;
+			}
+			else if (e instanceof IOException) {
+				throw (IOException) e;
+			}
 		}
 		return files;
 	}

spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -29,12 +29,12 @@ import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.SubscribableChannel;
-import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.transaction.TransactionDefinition;
@@ -81,8 +81,8 @@ public class FileInboundTransactionTests {
 		final AtomicBoolean crash = new AtomicBoolean();
 		input.subscribe(new MessageHandler() {
 
+			@Override
 			public void handleMessage(Message<?> message) throws MessagingException {
-				System.out.println(message);
 				if (crash.get()) {
 					throw new MessagingException("eek");
 				}
@@ -96,14 +96,12 @@ public class FileInboundTransactionTests {
 		Message<?> result = successChannel.receive(10000);
 		assertNotNull(result);
 		assertEquals(Boolean.TRUE, result.getPayload());
-		System.out.println(result);
 		assertFalse(file.delete());
 		crash.set(true);
 		file = new File(tmpDir + "/si-test1/bar");
 		file.createNewFile();
 		result = failureChannel.receive(10000);
 		assertNotNull(result);
-		System.out.println(result);
 		assertTrue(file.delete());
 		assertEquals("foo", result.getPayload());
 		pseudoTx.stop();
@@ -117,8 +115,8 @@ public class FileInboundTransactionTests {
 		final AtomicBoolean crash = new AtomicBoolean();
 		txInput.subscribe(new MessageHandler() {
 
+			@Override
 			public void handleMessage(Message<?> message) throws MessagingException {
-				System.out.println(message);
 				if (crash.get()) {
 					throw new MessagingException("eek");
 				}
@@ -133,14 +131,12 @@ public class FileInboundTransactionTests {
 		assertNotNull(result);
 		assertEquals(Boolean.TRUE, result.getPayload());
 		assertTrue(file.delete());
-		System.out.println(result);
 		assertTrue(transactionManager.getCommitted());
 		crash.set(true);
 		file = new File(tmpDir + "/si-test2/qux");
 		file.createNewFile();
 		result = failureChannel.receive(10000);
 		assertNotNull(result);
-		System.out.println(result);
 		assertTrue(file.delete());
 		assertEquals(Boolean.TRUE, result.getPayload());
 		realTx.stop();

spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java

@@ -109,7 +109,6 @@ public class FileReadingMessageSourceIntegrationTests {
 	@Test
 	public void getFiles() throws Exception {
 		Message<File> received1 = pollableFileSource.receive();
-		System.out.println("receive files round 1");
 		assertNotNull("This should return the first message", received1);
 		pollableFileSource.onSend(received1);
 		Message<File> received2 = pollableFileSource.receive();

spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013 the original author or authors.
+ * Copyright 2013-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -103,7 +103,6 @@ public class FileReadingMessageSourcePersistentFilterIntegrationTests {
 	@Test
 	public void getFiles() throws Exception {
 		Message<File> received1 = pollableFileSource.receive();
-		System.out.println("receive files round 1");
 		assertNotNull("This should return the first message", received1);
 		pollableFileSource.onSend(received1);
 		Message<File> received2 = pollableFileSource.receive();

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java

@@ -27,13 +27,20 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 
 import java.io.ByteArrayOutputStream;
 import java.io.File;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Arrays;
+import java.util.Calendar;
 import java.util.List;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.commons.net.ftp.FTPFile;
@@ -41,6 +48,9 @@ import org.hamcrest.Matchers;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -55,11 +65,14 @@ import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.TestFtpServer;
 import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.PartialSuccessException;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.annotation.DirtiesContext.ClassMode;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.FileCopyUtils;
@@ -72,6 +85,7 @@ import org.springframework.util.FileCopyUtils;
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD)
 public class FtpServerOutboundTests {
 
 	@Autowired
@@ -376,6 +390,137 @@ public class FtpServerOutboundTests {
 				TestUtils.getPropertyValue(result.getHeaders().get(FileHeaders.REMOTE_SESSION), "targetSession"));
 	}
 
+	@Test
+	public void testMgetPartial() throws Exception {
+		Session<FTPFile> session = spyOnSession();
+		doAnswer(new Answer<String[]>() {
+
+			@Override
+			public String[] answer(InvocationOnMock invocation) throws Throwable {
+				String[] files = (String[]) invocation.callRealMethod();
+				// add an extra file where the get will fail
+				files = Arrays.copyOf(files, files.length + 1);
+				files[files.length - 1] = "bogus.txt";
+				return files;
+			}
+		}).when(session).listNames("ftpSource/subFtpSource/*");
+		String dir = "ftpSource/subFtpSource/";
+		try {
+			this.inboundMGet.send(new GenericMessage<Object>(dir + "*"));
+			fail("expected exception");
+		}
+		catch (PartialSuccessException e) {
+			assertEquals(2, e.getDerivedInput().size());
+			assertEquals(1, e.getPartialResults().size());
+			assertThat(e.getCause().getMessage(),
+					containsString("/ftpSource/subFtpSource/bogus.txt: No such file or directory."));
+		}
+
+	}
+
+	@Test
+	public void testMgetRecursivePartial() throws Exception {
+		Session<FTPFile> session = spyOnSession();
+		doAnswer(new Answer<FTPFile[]>() {
+
+			@Override
+			public FTPFile[] answer(InvocationOnMock invocation) throws Throwable {
+				FTPFile[] files = (FTPFile[]) invocation.callRealMethod();
+				// add an extra file where the get will fail
+				files = Arrays.copyOf(files, files.length + 1);
+				FTPFile bogusFile = new FTPFile();
+				bogusFile.setName("bogus.txt");
+				bogusFile.setTimestamp(Calendar.getInstance());
+				files[files.length - 1] = bogusFile;
+				return files;
+			}
+		}).when(session).list("ftpSource/subFtpSource/");
+		String dir = "ftpSource/";
+		try {
+			this.inboundMGetRecursive.send(new GenericMessage<Object>(dir + "*"));
+			fail("expected exception");
+		}
+		catch (PartialSuccessException e) {
+			assertEquals(4, e.getDerivedInput().size());
+			assertEquals(2, e.getPartialResults().size());
+			assertThat(e.getCause().getMessage(),
+					containsString("/ftpSource/subFtpSource/bogus.txt: No such file or directory."));
+		}
+	}
+
+	@Test
+	public void testMputPartial() throws Exception {
+		Session<FTPFile> session = spyOnSession();
+		doAnswer(new Answer<Void>() {
+
+			@Override
+			public Void answer(InvocationOnMock invocation) throws Throwable {
+				throw new IOException("Failed to send localSource2");
+			}
+
+		}).when(session).write(Mockito.any(InputStream.class), Mockito.contains("localSource2"));
+		try {
+			this.inboundMPut.send(new GenericMessage<File>(this.ftpServer.getSourceLocalDirectory()));
+			fail("expected exception");
+		}
+		catch (PartialSuccessException e) {
+			assertEquals(3, e.getDerivedInput().size());
+			assertEquals(1, e.getPartialResults().size());
+			assertEquals("ftpTarget/localSource1.txt", e.getPartialResults().iterator().next());
+			assertThat(e.getCause().getMessage(),
+					containsString("Failed to send localSource2"));
+		}
+	}
+
+	@Test
+	public void testMputRecursivePartial() throws Exception {
+		Session<FTPFile> session = spyOnSession();
+		File sourceLocalSubDirectory =  new File(ftpServer.getSourceLocalDirectory(), "subLocalSource");
+		assertTrue(sourceLocalSubDirectory.isDirectory());
+		File extra = new File(sourceLocalSubDirectory, "subLocalSource2.txt");
+		FileOutputStream writer = new FileOutputStream(extra);
+		writer.write("foo".getBytes());
+		writer.close();
+		doAnswer(new Answer<Void>() {
+
+			@Override
+			public Void answer(InvocationOnMock invocation) throws Throwable {
+				throw new IOException("Failed to send subLocalSource2");
+			}
+
+		}).when(session).write(Mockito.any(InputStream.class), Mockito.contains("subLocalSource2"));
+		try {
+			this.inboundMPutRecursive.send(new GenericMessage<File>(this.ftpServer.getSourceLocalDirectory()));
+			fail("expected exception");
+		}
+		catch (PartialSuccessException e) {
+			assertEquals(3, e.getDerivedInput().size());
+			assertEquals(2, e.getPartialResults().size());
+			assertThat(e.getCause(), Matchers.instanceOf(PartialSuccessException.class));
+			PartialSuccessException cause = (PartialSuccessException) e.getCause();
+			assertEquals(2, cause.getDerivedInput().size());
+			assertEquals(1, cause.getPartialResults().size());
+			assertThat(cause.getCause().getMessage(), containsString("Failed to send subLocalSource2"));
+		}
+		extra.delete();
+	}
+
+	private Session<FTPFile> spyOnSession() {
+		Session<FTPFile> session = spy(this.ftpSessionFactory.getSession());
+		session.close();
+		@SuppressWarnings("unchecked")
+		BlockingQueue<Session<FTPFile>> cache = TestUtils.getPropertyValue(ftpSessionFactory, "pool.available",
+				BlockingQueue.class);
+		assertNotNull(cache.poll());
+		cache.offer(session);
+		@SuppressWarnings("unchecked")
+		Set<Session<FTPFile>> allocated = TestUtils.getPropertyValue(ftpSessionFactory, "pool.allocated",
+				Set.class);
+		allocated.clear();
+		allocated.add(session);
+		return session;
+	}
+
 	private void assertLength6(FtpRemoteFileTemplate template) {
 		FTPFile[] files = template.execute(new SessionCallback<FTPFile, FTPFile[]>() {
 

src/reference/asciidoc/ftp.adoc

@@ -398,6 +398,8 @@ The `-dirs` option is not allowed (the recursive mget uses the recursive `ls` to
 Typically, you would use the `#remoteDirectory` variable in the `local-directory-expression` so that the remote directory structure is retained locally.
 =====
 
+See also <<ftp-partial>>.
+
 *put*
 
 _put_ sends a file to the remote server; the payload of the message can be a `java.io.File`, a `byte[]` or a `String`.
@@ -422,6 +424,8 @@ Subdirectories that do not pass the filter are not recursed.
 
 The message payload resulting from an _mget_ operation is a `List<String>` object - a List of remote file paths resulting from the transfer.
 
+See also <<ftp-partial>>.
+
 *rm*
 
 The _rm_ command has no options.
@@ -480,6 +484,40 @@ file exists (`PUT` and `MPUT`). Supported modes are `REPLACE`, `APPEND`, `FAIL`
 For backwards compatibility, the default mode for `PUT` and `MPUT` operations is `REPLACE` and for `GET` and `MGET`
 operations, the default is `FAIL`.
 
+[[ftp-partial]]
+==== Outbound Gateway Partial Success (mget and mput)
+
+When performing operations on multiple files (`mget` and `mput`) it is possible that an exception occurs some time after
+one or more files have been transferred.
+In this case (starting with _version 4.2_), a `PartialSuccessException` is thrown.
+As well as the usual `MessagingException` properties (`failedMessage` and `cause`), this exception has two additional
+properties:
+
+- `partialResults` - the successful transfer results.
+- `derivedInput` - the list of files generated from the request message (e.g. local files to transfer for an `mput`).
+
+This will enable you to determine which files were successfully transferred, and which were not.
+
+In the case of a recursive `mput`, the `PartialSuccessException` may have nested `PartialSuccessException` s.
+
+Consider:
+
+[source]
+----
+root/
+|- file1.txt
+|- subdir/
+   | - file2.txt
+   | - file3.txt
+|- zoo.txt
+----
+
+If the exception occurs on `file3.txt`, the `PartialSuccessException` thrown by the gateway will have `derivedInput`
+of `file1.txt`, `subdir`, `zoo.txt` and `partialResults` of `file1.txt`.
+It's `cause` will be another `PartialSuccessException` with `derivedInput` of `file2.txt`, `file3.txt` and
+`partialResults` of `file2.txt`.
+
+
 [[ftp-session-caching]]
 === FTP Session Caching
 

src/reference/asciidoc/sftp.adoc

@@ -462,6 +462,8 @@ The `-dirs` option is not allowed (the recursive mget uses the recursive `ls` to
 Typically, you would use the `#remoteDirectory` variable in the `local-directory-expression` so that the remote directory structure is retained locally.
 =====
 
+See also <<sftp-partial>>
+
 *put*
 
 _put_ sends a file to the remote server; the payload of the message can be a `java.io.File`, a `byte[]` or a `String`.
@@ -486,6 +488,8 @@ Subdirectories that do not pass the filter are not recursed.
 
 The message payload resulting from an _mget_ operation is a `List<String>` object - a List of remote file paths resulting from the transfer.
 
+See also <<sftp-partial>>
+
 *rm*
 
 The _rm_ command has no options.
@@ -544,6 +548,39 @@ file exists (`PUT` and `MPUT`). Supported modes are `REPLACE`, `APPEND`, `FAIL`
 For backwards compatibility, the default mode for `PUT` and `MPUT` operations is `REPLACE` and for `GET` and `MGET`
 operations, the default is `FAIL`.
 
+[[sftp-partial]]
+==== Outbound Gateway Partial Success (mget and mput)
+
+When performing operations on multiple files (`mget` and `mput`) it is possible that an exception occurs some time after
+one or more files have been transferred.
+In this case (starting with _version 4.2_), a `PartialSuccessException` is thrown.
+As well as the usual `MessagingException` properties (`failedMessage` and `cause`), this exception has two additional
+properties:
+
+- `partialResults` - the successful transfer results.
+- `derivedInput` - the list of files generated from the request message (e.g. local files to transfer for an `mput`).
+
+This will enable you to determine which files were successfully transferred, and which were not.
+
+In the case of a recursive `mput`, the `PartialSuccessException` may have nested `PartialSuccessException` s.
+
+Consider:
+
+[source]
+----
+root/
+|- file1.txt
+|- subdir/
+   | - file2.txt
+   | - file3.txt
+|- zoo.txt
+----
+
+If the exception occurs on `file3.txt`, the `PartialSuccessException` thrown by the gateway will have `derivedInput`
+of `file1.txt`, `subdir`, `zoo.txt` and `partialResults` of `file1.txt`.
+It's `cause` will be another `PartialSuccessException` with `derivedInput` of `file2.txt`, `file3.txt` and
+`partialResults` of `file2.txt`.
+
 [[sftp-jsch-logging]]
 === SFTP/JSCH Logging
 

src/reference/asciidoc/whats-new.adoc

@@ -241,6 +241,15 @@ until they all arrive, and are then released individually. See <<aggregator>> fo
 
 ==== (S)FTP Changes
 
+===== Inbound channel adapters
+
 You can now specify a `remote-directory-expression` on the inbound channel adapters, to determine the directory
 at runtime.
 See <<ftp>> and <<sftp>> for more information.
+
+===== Gateway Partial Results
+
+When use FTP/SFTP outbound gateways to operate on multiple files (`mget`, `mput`), it is possible for an exception to
+occur after part of the request is completed.
+If such a condition occurs, a `PartialSuccessException` is thrown containing the partial results.
+See <<ftp-outbound-gateway>> and <<sftp-outbound-gateway>> for more information.