From a38bf603324eee05f09ab943147cc359425fa86a Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 19 Jun 2017 10:38:24 -0400 Subject: [PATCH] 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** --- .../file/FileWritingMessageHandler.java | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index a7ee18f883..8b462a5c5d 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/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(); - state.close(); - if (FileWritingMessageHandler.this.logger.isDebugEnabled()) { - FileWritingMessageHandler.this.logger.debug("Flushed: " + entry.getKey()); + if (state.close()) { + if (FileWritingMessageHandler.this.logger.isDebugEnabled()) { + FileWritingMessageHandler.this.logger.debug("Flushed: " + entry.getKey()); + } + iterator.remove(); + } + else { + break; // interrupted } } }