diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
index e0f38ab5c0..13a7e966fd 100644
--- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
+++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
@@ -27,7 +27,6 @@ import java.util.concurrent.PriorityBlockingQueue;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.core.io.Resource;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessagingException;
@@ -39,16 +38,19 @@ import org.springframework.util.Assert;
 /**
  * PollableSource that creates messages from a file system directory. To prevent
  * messages for certain files, you may supply a {@link FileListFilter}. By
- * default, an {@link AcceptOnceFileListFilter} is used. It ensures files are picked
- * up only once from the directory.
- * <p>
+ * default, an {@link AcceptOnceFileListFilter} is used. It ensures files are
+ * picked up only once from the directory.
+ * <p/>
  * A common problem with reading files is that a file may be detected before it
- * is ready. The default {@link AcceptOnceFileListFilter} does not prevent this. In
- * most cases, this can be prevented if the file-writing process renames each
+ * is ready. The default {@link AcceptOnceFileListFilter} does not prevent this.
+ * In most cases, this can be prevented if the file-writing process renames each
  * file as soon as it is ready for reading. A pattern-matching filter that
  * accepts only files that are ready (e.g. based on a known suffix), composed
- * with the default {@link AcceptOnceFileListFilter} would allow for this.
- * See {@ link CompositeFileFilter} for a way to do this.
+ * with the default {@link AcceptOnceFileListFilter} would allow for this. See
+ * {@link CompositeFileFilter} for a way to do this.
+ * <p/>
+ * FileReadingMessageSource is fully thread-safe under concurrent receives and
+ * message delivery callbacks.
  * 
  * @author Iwein Fuld
  */
@@ -58,11 +60,15 @@ public class FileReadingMessageSource implements MessageSource<File>, MessageDel
 
 	private volatile File inputDirectory;
 
+	/**
+	 * {@link PriorityBlockingQueue#iterator()} throws
+	 * {@link ConcurrentModificationException} in Java 5. There is no locking
+	 * around the queue, so there is also no iteration.
+	 */
 	private final Queue<File> toBeReceived = new PriorityBlockingQueue<File>();
 
 	private volatile FileListFilter filter = new AcceptOnceFileListFilter();
 
-
 	public void setInputDirectory(Resource inputDirectory) {
 		Assert.notNull(inputDirectory, "inputDirectory cannot be null");
 		Assert.isTrue(inputDirectory.exists(), inputDirectory + " doesn't exist.");
@@ -80,7 +86,8 @@ public class FileReadingMessageSource implements MessageSource<File>, MessageDel
 	 * {@link AcceptOnceFileListFilter} with no bounds is used. In most cases a
 	 * customized {@link FileFilter} will be needed to deal with modification
 	 * and duplication concerns. If multiple filters are required a
-	 * {@link CompositeFileListFilter} can be used to group them together <p/>
+	 * {@link CompositeFileListFilter} can be used to group them together
+	 * <p/>
 	 * <b>Note that the supplied filter must be thread safe</b>.
 	 */
 	public void setFilter(FileListFilter filter) {
@@ -106,8 +113,6 @@ public class FileReadingMessageSource implements MessageSource<File>, MessageDel
 		List<File> filteredFiles = filter.filterFiles((inputDirectory.listFiles()));
 		Set<File> freshFiles = new HashSet<File>(filteredFiles);
 		if (!freshFiles.isEmpty()) {
-			// don't duplicate what's on the queue already
-			freshFiles.removeAll(toBeReceived);
 			toBeReceived.addAll(freshFiles);
 			if (logger.isDebugEnabled()) {
 				logger.debug("Added to queue: " + freshFiles);
@@ -116,17 +121,19 @@ public class FileReadingMessageSource implements MessageSource<File>, MessageDel
 	}
 
 	/**
-	 * In concurrent scenarios onFailure() might cause failing files to be
-	 * ignored. If this is not acceptable access to this method should be
-	 * synchronized on this instance externally.
+	 * Adds the failed message back to the 'toBeReceived' queue.
 	 */
 	public void onFailure(Message<File> failedMessage, Throwable t) {
 		if (logger.isWarnEnabled()) {
 			logger.warn("Failed to send: " + failedMessage);
 		}
-		toBeReceived.add(failedMessage.getPayload());
+		toBeReceived.offer(failedMessage.getPayload());
 	}
 
+	/**
+	 * The message is just logged. It was already removed from the queue during
+	 * the call to <code>receive()</code>
+	 */
 	public void onSend(Message<File> sentMessage) {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Sent: " + sentMessage);