INT-4271: FWMH: Fix Race Condition with stop()

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

The `Flusher.run()` is `synchronized` on the `FWMH`; a `stop()` while it's running will interrupt the task
which could cause a state to be removed without a flush actually happening (interrupt
on lock acquisition).

- synchronize `stop()` so it won't interrupt a running flusher
- detect an interrupt in the flusher and stop flushing


**Cherry-pick to 4.3.x**

spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java

@@ -474,7 +474,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 	}
 
 	@Override
-	public void stop() {
+	public synchronized void stop() {
 		if (this.flushTask != null) {
 			this.flushTask.cancel(true);
 			this.flushTask = null;
@@ -1005,7 +1005,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 			this.lock = lock;
 		}
 
-		private void close() {
+		private boolean close() {
 			try {
 				this.lock.lockInterruptibly();
 				try {
@@ -1019,9 +1019,11 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 				catch (IOException e) {
 					// ignore
 				}
+				return true;
 			}
 			catch (InterruptedException e1) {
 				Thread.currentThread().interrupt();
+				return false;
 			}
 			finally {
 				this.lock.unlock();
@@ -1047,10 +1049,14 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
 					FileState state = entry.getValue();
 					if (state.lastWrite < expired ||
 							(!FileWritingMessageHandler.this.flushWhenIdle && state.firstWrite < expired)) {
-						iterator.remove();
+						if (state.close()) {
-						state.close();
+							if (FileWritingMessageHandler.this.logger.isDebugEnabled()) {
-						if (FileWritingMessageHandler.this.logger.isDebugEnabled()) {
+								FileWritingMessageHandler.this.logger.debug("Flushed: " + entry.getKey());
-							FileWritingMessageHandler.this.logger.debug("Flushed: " + entry.getKey());
+							}
+							iterator.remove();
+						}
+						else {
+							break; // interrupted
 						}
 					}
 				}