diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java
index b814dff3b9..d8ea345797 100644
--- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java
+++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java
@@ -69,6 +69,8 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
 
 	private volatile File localWorkingDirectory;
 
+	private int maxFilesPerPayload = -1;
+
 	private final FTPClient client;
 
 	public FtpSource(MessageCreator<List<File>, List<File>> messageCreator) {
@@ -97,6 +99,11 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
 		this.password = password;
 	}
 
+	public void setMaxMessagesPerPayload(int maxMessagesPerPayload) {
+		Assert.isTrue(maxMessagesPerPayload > 0, "'maxMessagesPerPayload' should greater than 0");
+		this.maxFilesPerPayload = maxMessagesPerPayload;
+	}
+
 	public void setRemoteWorkingDirectory(String remoteWorkingDirectory) {
 		Assert.notNull(remoteWorkingDirectory, "'remoteWorkingDirectory' cannot be null");
 		// FtpClient is picky about "", so we make it happy
@@ -152,7 +159,12 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
 		establishConnection();
 		List<File> files = new ArrayList<File>();
 		Set<String> backlog = this.getDirectoryContentManager().getBacklog().keySet();
-		for (String fileName : backlog) {
+		int i = 0;
+		Iterator<String> iterator = backlog.iterator();
+		while (iterator.hasNext() && (maxFilesPerPayload == -1 || i < maxFilesPerPayload)) {
+			i++;
+			String fileName = iterator.next();
+
 			File file = new File(this.localWorkingDirectory, fileName);
 			if (file.exists()) {
 				file.delete();
@@ -186,6 +198,7 @@ public class FtpSource extends AbstractDirectorySource<List<File>> implements Di
 		disconnect();
 	}
 
+	@SuppressWarnings("unchecked")
 	@Override
 	public void onSend(Message<?> message) {
 		List<File> files = ((Message<List<File>>) message).getPayload();
diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java
index 01af3ded41..121e6908a8 100644
--- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java
+++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java
@@ -4,6 +4,8 @@ import static org.easymock.classextension.EasyMock.*;
 import static org.junit.Assert.*;
 
 import java.io.File;
+import java.io.FileFilter;
+import java.io.FilenameFilter;
 import java.io.OutputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -12,7 +14,10 @@ import java.util.List;
 
 import org.apache.commons.net.ftp.FTPClient;
 import org.apache.commons.net.ftp.FTPFile;
+import org.apache.oro.io.Perl5FilenameFilter;
 import org.easymock.IAnswer;
+import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -40,6 +45,8 @@ public class FtpSourceTests {
 	private FtpSource ftpSource;
 
 	private Long size = 100l;
+	
+	private File testDir = new File(this.getClass().getName());
 
 	@Before
 	public void initializeFtpSource() {
@@ -103,12 +110,12 @@ public class FtpSourceTests {
 		expect(ftpClient.printWorkingDirectory()).andReturn("/");
 
 		// get files
-		expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test", "test2")).times(2);
+		expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test1", "test2")).times(2);
 
-		expect(ftpClient.retrieveFile(eq("test"), isA(OutputStream.class))).andReturn(true);
+		expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
 		expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true);
 		// create message
-		List<File> files = Arrays.asList(new File("test"), new File("test2"));
+		List<File> files = Arrays.asList(new File("test1"), new File("test2"));
 		expect(messageCreator.createMessage(isA(List.class))).andReturn(new GenericMessage(files));
 		ftpClient.disconnect();
 
@@ -127,20 +134,20 @@ public class FtpSourceTests {
 		// assume client already connected
 		expect(ftpClient.isConnected()).andReturn(true).anyTimes();
 		// first run
-		FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test", "test2");
+		FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2");
 		expect(ftpClient.listFiles()).andReturn(mockedFTPFiles);
-		expect(ftpClient.retrieveFile(eq("test"), isA(OutputStream.class))).andReturn(true);
+		expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
 		expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true);
 
 		ftpClient.disconnect();
 		// second run, change the date so the messages should be retrieved again
 		// expect(ftpClient.isConnected()).andReturn(true);
-		FTPFile[] mockedFTPFiles2 = mockedFTPFilesNamed("test", "test2");
+		FTPFile[] mockedFTPFiles2 = mockedFTPFilesNamed("test1", "test2");
 		expect(ftpClient.listFiles()).andReturn(mockedFTPFiles2);
-		expect(ftpClient.retrieveFile(eq("test"), isA(OutputStream.class))).andReturn(true);
+		expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
 		expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true);
 		// create message
-		List<File> files = Arrays.asList(new File("test"), new File("test2"));
+		List<File> files = Arrays.asList(new File("test1"), new File("test2"));
 		expect(messageCreator.createMessage(isA(List.class))).andReturn(new GenericMessage(files)).times(2);
 		ftpClient.disconnect();
 
@@ -153,17 +160,15 @@ public class FtpSourceTests {
 	}
 
 	@Test
-	@Ignore
 	public void retrieveMaxFilesPerPayload() throws Exception {
 
-		// TODO: uncomment (and un-ignore) this when FtpSource is committed
-		//this.ftpSource.setMaxMessagesPerPayload(2);
+		 this.ftpSource.setMaxMessagesPerPayload(2);
 		// assume client already connected
 		expect(ftpClient.isConnected()).andReturn(true).anyTimes();
 		// first run
-		FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test", "test2", "test3");
+		FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2", "test3");
 		expect(ftpClient.listFiles()).andReturn(mockedFTPFiles);
-		expect(ftpClient.retrieveFile(eq("test"), isA(OutputStream.class))).andReturn(true);
+		expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
 		expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true);
 
 		ftpClient.disconnect();
@@ -189,4 +194,13 @@ public class FtpSourceTests {
 		assertEquals(1, receivedFiles2.getPayload().size());
 	}
 
+	@AfterClass
+	public static void deleteFiles() {
+		File file = new File("./");
+		File[] files = file.listFiles((FilenameFilter) new Perl5FilenameFilter("test\\d"));
+		for (File file2 : files) {
+			file2.delete();
+		}
+	}
+
 }