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**