Repository: hazelcast/hazelcast-simulator
Issue: Remote: when fail fast, test not aborting

When deliberately causing a test failure by killing the members (and having the load generated by a client),the testcase runner doesn't complete

```
INFO  14:49:10 AtomicLongTest Running 00d 00h 00m 20s (  6.67%)        54,600 ops     3,409.70 ops/s        292 µs (avg)      2,619 µs (99.9th)     19,628 µs (max)
INFO  14:49:15 Valid connection from /127.0.0.1:39036 (magic bytes found)
INFO  14:49:15 Killing working....
INFO  14:49:15 Kill send to worker [C_A2_W1]
INFO  14:49:20 AtomicLongTest Running 00d 00h 00m 30s ( 10.00%)        92,414 ops     3,781.40 ops/s        263 µs (avg)      1,140 µs (99.9th)     21,725 µs (max)
INFO  14:49:30 AtomicLongTest Running 00d 00h 00m 40s ( 13.33%)        92,915 ops        50.10 ops/s        295 µs (avg)      2,059 µs (99.9th)      2,772 µs (max)
INFO  14:49:40 AtomicLongTest Running 00d 00h 00m 50s ( 16.67%)       153,606 ops     6,069.10 ops/s        326 µs (avg)        332 µs (99.9th) 10,015,997 µs (max)
INFO  14:49:41 Valid connection from /127.0.0.1:39208 (magic bytes found)
INFO  14:49:41 Killing working....
INFO  14:49:41 Kill send to worker [C_A1_W1]
ERROR 14:49:47 Failure #1 C_A1_W3 run-AtomicLongTest WORKER_EXCEPTION[com.hazelcast.spi.exception.TargetDisconnectedException: Disconnecting from member [10.30.221.14]:5701 due to heartbeat problems. Current time: 2016-08-23 07:49:47.151. Last heartbeat requested: 2016-08-23 07:49:30.151. Last heartbeat received: 2016-08-23 07:49:30.153. Last read: 2016-08-23 07:49:46.307. Connection ClientConnection{live=false, connectionId=2, socketChannel=DefaultSocketChannelWrapper{socketChannel=java.nio.channels.SocketChannel[closed]}, remoteEndpoint=[10.30.221.14]:5701, lastReadTime=2016-08-23 07:49:46.307, lastWriteTime=2016-08-23 07:49:46.305, closedTime=2016-08-23 07:49:46.307, lastHeartbeatRequested=2016-08-23 07:49:30.151, lastHeartbeatReceived=2016-08-23 07:49:30.153}]
INFO  14:49:50 AtomicLongTest Running 00d 00h 01m 00s ( 20.00%)       195,126 ops     4,151.58 ops/s        145 µs (avg)        256 µs (99.9th)     27,918 µs (max)
INFO  14:49:50 AtomicLongTest Critical failure detected, aborting run phase
INFO  14:49:51 AtomicLongTest Test finished running
INFO  14:49:51 AtomicLongTest Executing Test stop
ERROR 14:52:51 Failure #2 C_A1_W3 WORKER_TIMEOUT[Worker has not sent a message for 180 seconds]
```

I guess it is waiting for the old member count, not the current member count.


Changed files:
simulator/src/main/java/com/hazelcast/simulator/agent/FailureSender.java
simulator/src/test/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitorTest.java
simulator/src/main/java/com/hazelcast/simulator/agent/FailureSenderImpl.java
simulator/src/test/java/com/hazelcast/simulator/agent/FailureSenderImplTest.java
simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessManager.java
simulator/src/main/java/com/hazelcast/simulator/agent/Agent.java
simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitor.java
simulator/src/main/java/com/hazelcast/simulator/coordinator/Coordinator.java
simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcess.java

diff --git a/simulator/src/main/java/com/hazelcast/simulator/agent/Agent.java b/simulator/src/main/java/com/hazelcast/simulator/agent/Agent.java
index 49616abb7..03c687d95 100644
--- a/simulator/src/main/java/com/hazelcast/simulator/agent/Agent.java
+++ b/simulator/src/main/java/com/hazelcast/simulator/agent/Agent.java
@@ -55,7 +55,7 @@ public class Agent {
     private final String cloudCredential;
 
     private final AgentConnector agentConnector;
-    private final FailureSenderImpl failureSender;
+    private final FailureHandlerImpl failureSender;
     private final WorkerProcessFailureMonitor workerProcessFailureMonitor;
 
     private volatile TestSuite testSuite;
@@ -74,7 +74,7 @@ public class Agent {
         this.cloudCredential = cloudCredential;
 
         this.agentConnector = AgentConnector.createInstance(this, workerProcessManager, port, threadPoolSize);
-        this.failureSender = new FailureSenderImpl(publicAddress, agentConnector);
+        this.failureSender = new FailureHandlerImpl(publicAddress, agentConnector);
         this.workerProcessFailureMonitor = new WorkerProcessFailureMonitor(failureSender, workerProcessManager,
                 workerLastSeenTimeoutSeconds);
 
diff --git a/simulator/src/main/java/com/hazelcast/simulator/agent/FailureSenderImpl.java b/simulator/src/main/java/com/hazelcast/simulator/agent/FailureHandlerImpl.java
similarity index 82%
rename from simulator/src/main/java/com/hazelcast/simulator/agent/FailureSenderImpl.java
rename to simulator/src/main/java/com/hazelcast/simulator/agent/FailureHandlerImpl.java
index 45f1f3fa7..e9b341ab7 100644
--- a/simulator/src/main/java/com/hazelcast/simulator/agent/FailureSenderImpl.java
+++ b/simulator/src/main/java/com/hazelcast/simulator/agent/FailureHandlerImpl.java
@@ -15,6 +15,7 @@
  */
 package com.hazelcast.simulator.agent;
 
+import com.hazelcast.simulator.agent.workerprocess.FailureHandler;
 import com.hazelcast.simulator.agent.workerprocess.WorkerProcess;
 import com.hazelcast.simulator.common.FailureType;
 import com.hazelcast.simulator.common.TestSuite;
@@ -30,9 +31,9 @@ import org.apache.log4j.Logger;
 import static com.hazelcast.simulator.protocol.core.SimulatorAddress.COORDINATOR;
 import static java.lang.String.format;
 
-class FailureSenderImpl implements FailureSender {
+class FailureHandlerImpl implements FailureHandler {
 
-    private static final Logger LOGGER = Logger.getLogger(FailureSenderImpl.class);
+    private static final Logger LOGGER = Logger.getLogger(FailureHandlerImpl.class);
 
     private final String agentAddress;
     private final AgentConnector agentConnector;
@@ -41,7 +42,7 @@ class FailureSenderImpl implements FailureSender {
 
     private int failureCount;
 
-    FailureSenderImpl(String agentAddress, AgentConnector agentConnector) {
+    FailureHandlerImpl(String agentAddress, AgentConnector agentConnector) {
         this.agentAddress = agentAddress;
         this.agentConnector = agentConnector;
     }
@@ -51,14 +52,26 @@ class FailureSenderImpl implements FailureSender {
     }
 
     @Override
-    public boolean sendFailureOperation(String message,
-                                        FailureType type,
-                                        WorkerProcess workerProcess,
-                                        String testId,
-                                        String cause) {
-        boolean sentSuccessfully = true;
+    public boolean handle(String message, FailureType type, WorkerProcess workerProcess, String testId, String cause) {
+        boolean send = true;
+
+        if (!workerProcess.isFailureIgnored()) {
+            send = send(message, type, workerProcess, testId, cause);
+        }
+
+        if (type.isWorkerFinishedFailure()) {
+            SimulatorAddress workerAddress = workerProcess.getAddress();
+            unblockPendingFutures(workerAddress);
+            removeFinishedWorker(workerAddress, type);
+        }
+
+        return send;
+    }
+
+    private boolean send(String message, FailureType type, WorkerProcess workerProcess, String testId, String cause) {
         SimulatorAddress workerAddress = workerProcess.getAddress();
         String workerId = workerProcess.getId();
+
         FailureOperation operation = new FailureOperation(message, type, workerAddress, agentAddress,
                 workerProcess.getHazelcastAddress(), workerId, testId, testSuite, cause);
 
@@ -69,28 +82,23 @@ class FailureSenderImpl implements FailureSender {
                     operation.getLogMessage(++failureCount)));
         }
 
+        boolean send = true;
         try {
             Response response = agentConnector.write(COORDINATOR, operation);
             ResponseType firstErrorResponseType = response.getFirstErrorResponseType();
             if (firstErrorResponseType != ResponseType.SUCCESS) {
                 LOGGER.error(format("Could not send failure to coordinator: %s", firstErrorResponseType));
-                sentSuccessfully = false;
+                send = false;
             } else if (!type.isPoisonPill()) {
                 LOGGER.info("Failure successfully sent to Coordinator!");
             }
         } catch (SimulatorProtocolException e) {
             if (!(e.getCause() instanceof InterruptedException)) {
                 LOGGER.error(format("Could not send failure to coordinator! %s", operation.getFileMessage()), e);
-                sentSuccessfully = false;
+                send = false;
             }
         }
-
-        if (type.isWorkerFinishedFailure()) {
-            unblockPendingFutures(workerAddress);
-            removeFinishedWorker(workerAddress, type);
-        }
-
-        return sentSuccessfully;
+        return send;
     }
 
     private void unblockPendingFutures(SimulatorAddress workerAddress) {
diff --git a/simulator/src/main/java/com/hazelcast/simulator/agent/FailureSender.java b/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/FailureHandler.java
similarity index 72%
rename from simulator/src/main/java/com/hazelcast/simulator/agent/FailureSender.java
rename to simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/FailureHandler.java
index 3f18e7884..d5a54d5e4 100644
--- a/simulator/src/main/java/com/hazelcast/simulator/agent/FailureSender.java
+++ b/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/FailureHandler.java
@@ -13,12 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.hazelcast.simulator.agent;
+package com.hazelcast.simulator.agent.workerprocess;
 
-import com.hazelcast.simulator.agent.workerprocess.WorkerProcess;
 import com.hazelcast.simulator.common.FailureType;
 
-public interface FailureSender {
+public interface FailureHandler {
 
-    boolean sendFailureOperation(String message, FailureType type, WorkerProcess workerProcess, String testId, String cause);
+    boolean handle(String message, FailureType type, WorkerProcess workerProcess, String testId, String cause);
 }
diff --git a/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcess.java b/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcess.java
index 6e25e4a13..cd2892aec 100644
--- a/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcess.java
+++ b/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcess.java
@@ -33,6 +33,7 @@ public class WorkerProcess {
     private volatile boolean isFinished;
     private volatile Process process;
     private volatile String hzAddress;
+    private volatile boolean failureIgnored;
 
     public WorkerProcess(SimulatorAddress address, String id, File workerHome) {
         this.address = address;
@@ -95,4 +96,12 @@ public class WorkerProcess {
     public void setHzAddress(String memberAddress) {
         this.hzAddress = memberAddress;
     }
+
+    public boolean isFailureIgnored() {
+        return failureIgnored;
+    }
+
+    public void setFailureIgnored(boolean failureIgnored) {
+        this.failureIgnored = failureIgnored;
+    }
 }
diff --git a/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitor.java b/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitor.java
index 0c7db04c2..73d30dc3e 100644
--- a/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitor.java
+++ b/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitor.java
@@ -15,13 +15,11 @@
  */
 package com.hazelcast.simulator.agent.workerprocess;
 
-import com.hazelcast.simulator.agent.FailureSender;
 import com.hazelcast.simulator.common.FailureType;
 import org.apache.log4j.Logger;
 
 import java.io.File;
 import java.io.FilenameFilter;
-import java.util.concurrent.TimeUnit;
 
 import static com.hazelcast.simulator.common.FailureType.WORKER_EXCEPTION;
 import static com.hazelcast.simulator.common.FailureType.WORKER_EXIT;
@@ -34,6 +32,7 @@ import static com.hazelcast.simulator.utils.FileUtils.fileAsText;
 import static com.hazelcast.simulator.utils.FileUtils.rename;
 import static com.hazelcast.simulator.utils.FormatUtils.NEW_LINE;
 import static java.lang.String.format;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 public class WorkerProcessFailureMonitor {
@@ -44,17 +43,17 @@ public class WorkerProcessFailureMonitor {
 
     private final MonitorThread monitorThread;
 
-    public WorkerProcessFailureMonitor(FailureSender failureSender,
+    public WorkerProcessFailureMonitor(FailureHandler failureHandler,
                                        WorkerProcessManager workerProcessManager,
                                        int lastSeenTimeoutSeconds) {
-        this(failureSender, workerProcessManager, lastSeenTimeoutSeconds, DEFAULT_CHECK_INTERVAL_MILLIS);
+        this(failureHandler, workerProcessManager, lastSeenTimeoutSeconds, DEFAULT_CHECK_INTERVAL_MILLIS);
     }
 
-    WorkerProcessFailureMonitor(FailureSender failureSender,
+    WorkerProcessFailureMonitor(FailureHandler failureHandler,
                                 WorkerProcessManager workerProcessManager,
                                 int lastSeenTimeoutSeconds,
                                 int checkIntervalMillis) {
-        monitorThread = new MonitorThread(failureSender, workerProcessManager, lastSeenTimeoutSeconds, checkIntervalMillis);
+        monitorThread = new MonitorThread(failureHandler, workerProcessManager, lastSeenTimeoutSeconds, checkIntervalMillis);
     }
 
     public void start() {
@@ -83,7 +82,7 @@ public class WorkerProcessFailureMonitor {
 
     private final class MonitorThread extends Thread {
 
-        private final FailureSender failureSender;
+        private final FailureHandler failureHandler;
         private final WorkerProcessManager workerProcessManager;
         private final int lastSeenTimeoutSeconds;
         private final int checkIntervalMillis;
@@ -91,14 +90,14 @@ public class WorkerProcessFailureMonitor {
         private volatile boolean running = true;
         private volatile boolean detectTimeouts;
 
-        private MonitorThread(FailureSender failureSender,
+        private MonitorThread(FailureHandler failureHandler,
                               WorkerProcessManager workerProcessManager,
                               int lastSeenTimeoutSeconds,
                               int checkIntervalMillis) {
             super("WorkerJvmFailureMonitorThread");
             setDaemon(true);
 
-            this.failureSender = failureSender;
+            this.failureHandler = failureHandler;
             this.workerProcessManager = workerProcessManager;
             this.lastSeenTimeoutSeconds = lastSeenTimeoutSeconds;
             this.checkIntervalMillis = checkIntervalMillis;
@@ -161,7 +160,7 @@ public class WorkerProcessFailureMonitor {
                 }
 
                 // we delete or rename the exception file so that we don't detect the same exception again
-                boolean send = failureSender.sendFailureOperation(
+                boolean send = failureHandler.handle(
                         "Worked ran into an unhandled exception", WORKER_EXCEPTION, workerProcess, testId, cause);
 
                 if (send) {
@@ -199,7 +198,7 @@ public class WorkerProcessFailureMonitor {
                 return;
             }
 
-            long elapsed = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - workerProcess.getLastSeen());
+            long elapsed = MILLISECONDS.toSeconds(System.currentTimeMillis() - workerProcess.getLastSeen());
             if (elapsed > 0 && elapsed % lastSeenTimeoutSeconds == 0) {
                 sendFailureOperation(
                         format("Worker has not sent a message for %d seconds", elapsed), WORKER_TIMEOUT, workerProcess);
@@ -229,7 +228,7 @@ public class WorkerProcessFailureMonitor {
         }
 
         private void sendFailureOperation(String message, FailureType type, WorkerProcess workerProcess) {
-            failureSender.sendFailureOperation(message, type, workerProcess, null, null);
+            failureHandler.handle(message, type, workerProcess, null, null);
         }
     }
 
diff --git a/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessManager.java b/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessManager.java
index a02cf8721..6f45b351c 100644
--- a/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessManager.java
+++ b/simulator/src/main/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessManager.java
@@ -44,8 +44,10 @@ public class WorkerProcessManager {
     }
 
     public void ignore(SimulatorAddress workerAddress) {
-        LOGGER.info("Dropping worker [" + workerAddress + "]");
-        workerProcesses.remove(workerAddress);
+        WorkerProcess workerProcess = workerProcesses.get(workerAddress);
+        if (workerProcess != null) {
+            workerProcess.setFailureIgnored(true);
+        }
     }
 
     public Collection<WorkerProcess> getWorkerProcesses() {
diff --git a/simulator/src/main/java/com/hazelcast/simulator/coordinator/Coordinator.java b/simulator/src/main/java/com/hazelcast/simulator/coordinator/Coordinator.java
index a5517d049..c72eca66c 100644
--- a/simulator/src/main/java/com/hazelcast/simulator/coordinator/Coordinator.java
+++ b/simulator/src/main/java/com/hazelcast/simulator/coordinator/Coordinator.java
@@ -428,9 +428,13 @@ public final class Coordinator {
         }
 
         SimulatorAddress memberAddress = randomMember.getAddress();
+
         componentRegistry.removeWorker(memberAddress);
+
         coordinatorConnector.write(memberAddress.getParent(), new IgnoreWorkerFailureOperation(memberAddress));
+
         coordinatorConnector.writeAsync(memberAddress, new KillWorkerOperation());
+
         LOGGER.info("Kill send to worker [" + memberAddress + "]");
     }
 
diff --git a/simulator/src/test/java/com/hazelcast/simulator/agent/FailureSenderImplTest.java b/simulator/src/test/java/com/hazelcast/simulator/agent/FailureHandlerImplTest.java
similarity index 84%
rename from simulator/src/test/java/com/hazelcast/simulator/agent/FailureSenderImplTest.java
rename to simulator/src/test/java/com/hazelcast/simulator/agent/FailureHandlerImplTest.java
index 82b74422a..d16d749d5 100644
--- a/simulator/src/test/java/com/hazelcast/simulator/agent/FailureSenderImplTest.java
+++ b/simulator/src/test/java/com/hazelcast/simulator/agent/FailureHandlerImplTest.java
@@ -31,10 +31,10 @@ import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-public class FailureSenderImplTest {
+public class FailureHandlerImplTest {
 
     private static final String FAILURE_MESSAGE = "failure message";
-    private static final String SESSION_ID = "FailureSenderImplTest";
+    private static final String SESSION_ID = "FailureHandlerImplTest";
     private static final String CAUSE = "any stacktrace";
 
     private SimulatorAddress workerAddress;
@@ -42,7 +42,7 @@ public class FailureSenderImplTest {
 
     private AgentConnector agentConnector;
 
-    private FailureSenderImpl failureSender;
+    private FailureHandlerImpl failureSender;
 
     private ResponseFuture responseFuture;
 
@@ -64,13 +64,13 @@ public class FailureSenderImplTest {
 
         TestSuite testSuite = new TestSuite();
 
-        failureSender = new FailureSenderImpl("127.0.0.1", agentConnector);
+        failureSender = new FailureHandlerImpl("127.0.0.1", agentConnector);
         failureSender.setTestSuite(testSuite);
     }
 
     @Test
     public void testSendFailureOperation() {
-        boolean success = failureSender.sendFailureOperation(FAILURE_MESSAGE, WORKER_EXCEPTION, workerProcess, SESSION_ID, CAUSE);
+        boolean success = failureSender.handle(FAILURE_MESSAGE, WORKER_EXCEPTION, workerProcess, SESSION_ID, CAUSE);
 
         assertTrue(success);
         assertFalse(responseFuture.isDone());
@@ -78,7 +78,7 @@ public class FailureSenderImplTest {
 
     @Test(timeout = 10000)
     public void testSendFailureOperation_whenWorkerIsFinished_thenUnblockResponseFutureByFailure() throws Exception {
-        boolean success = failureSender.sendFailureOperation(FAILURE_MESSAGE, WORKER_FINISHED, workerProcess, SESSION_ID, CAUSE);
+        boolean success = failureSender.handle(FAILURE_MESSAGE, WORKER_FINISHED, workerProcess, SESSION_ID, CAUSE);
 
         assertTrue(success);
         assertTrue(responseFuture.isDone());
@@ -90,7 +90,7 @@ public class FailureSenderImplTest {
         Response failureResponse = new Response(1, COORDINATOR, workerAddress, FAILURE_COORDINATOR_NOT_FOUND);
         when(agentConnector.write(any(SimulatorAddress.class), any(SimulatorOperation.class))).thenReturn(failureResponse);
 
-        boolean success = failureSender.sendFailureOperation(FAILURE_MESSAGE, WORKER_EXCEPTION, workerProcess, SESSION_ID, CAUSE);
+        boolean success = failureSender.handle(FAILURE_MESSAGE, WORKER_EXCEPTION, workerProcess, SESSION_ID, CAUSE);
 
         assertFalse(success);
         assertFalse(responseFuture.isDone());
@@ -101,7 +101,7 @@ public class FailureSenderImplTest {
         when(agentConnector.write(any(SimulatorAddress.class), any(SimulatorOperation.class)))
                 .thenThrow(new SimulatorProtocolException("expected exception"));
 
-        boolean success = failureSender.sendFailureOperation(FAILURE_MESSAGE, NETTY_EXCEPTION, workerProcess, SESSION_ID, CAUSE);
+        boolean success = failureSender.handle(FAILURE_MESSAGE, NETTY_EXCEPTION, workerProcess, SESSION_ID, CAUSE);
 
         assertFalse(success);
         assertFalse(responseFuture.isDone());
diff --git a/simulator/src/test/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitorTest.java b/simulator/src/test/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitorTest.java
index 4d1d8aee8..582470d42 100644
--- a/simulator/src/test/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitorTest.java
+++ b/simulator/src/test/java/com/hazelcast/simulator/agent/workerprocess/WorkerProcessFailureMonitorTest.java
@@ -1,6 +1,5 @@
 package com.hazelcast.simulator.agent.workerprocess;
 
-import com.hazelcast.simulator.agent.FailureSender;
 import com.hazelcast.simulator.common.FailureType;
 import com.hazelcast.simulator.protocol.core.Response;
 import com.hazelcast.simulator.protocol.core.SimulatorAddress;
@@ -57,7 +56,7 @@ public class WorkerProcessFailureMonitorTest {
 
     private static int addressIndex;
 
-    private FailureSender failureSender;
+    private FailureHandler failureHandler;
     private WorkerProcessManager workerProcessManager;
 
 
@@ -70,15 +69,15 @@ public class WorkerProcessFailureMonitorTest {
         simulatorHome = setupFakeEnvironment();
         workersHome = new File(simulatorHome, "workers");
 
-        failureSender = mock(FailureSender.class);
-        when(failureSender.sendFailureOperation(
+        failureHandler = mock(FailureHandler.class);
+        when(failureHandler.handle(
                 anyString(), any(FailureType.class), any(WorkerProcess.class), any(String.class), any(String.class)))
                 .thenReturn(true);
 
         workerProcessManager = new WorkerProcessManager();
 
         workerProcessFailureMonitor = new WorkerProcessFailureMonitor(
-                failureSender,
+                failureHandler,
                 workerProcessManager,
                 DEFAULT_LAST_SEEN_TIMEOUT_SECONDS,
                 DEFAULT_CHECK_INTERVAL);
@@ -94,18 +93,18 @@ public class WorkerProcessFailureMonitorTest {
 
     @Test
     public void testConstructor() {
-        workerProcessFailureMonitor = new WorkerProcessFailureMonitor(failureSender, workerProcessManager,
+        workerProcessFailureMonitor = new WorkerProcessFailureMonitor(failureHandler, workerProcessManager,
                 DEFAULT_LAST_SEEN_TIMEOUT_SECONDS);
         workerProcessFailureMonitor.start();
 
-        verifyZeroInteractions(failureSender);
+        verifyZeroInteractions(failureHandler);
     }
 
     @Test
     public void testRun_shouldSendNoFailures() {
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        verifyZeroInteractions(failureSender);
+        verifyZeroInteractions(failureHandler);
     }
 
     @Test(timeout = DEFAULT_TIMEOUT)
@@ -123,7 +122,7 @@ public class WorkerProcessFailureMonitorTest {
         assertTrueEventually(new AssertTask() {
             @Override
             public void run() throws Exception {
-                assertFailureType(failureSender, WORKER_OOM);
+                assertFailureType(failureHandler, WORKER_OOM);
             }
         });
     }
@@ -132,7 +131,7 @@ public class WorkerProcessFailureMonitorTest {
     public void testRun_shouldContinueAfterErrorResponse() {
         Response failOnceResponse = mock(Response.class);
         when(failOnceResponse.getFirstErrorResponseType()).thenReturn(FAILURE_COORDINATOR_NOT_FOUND).thenReturn(SUCCESS);
-        when(failureSender.sendFailureOperation(
+        when(failureHandler.handle(
                 anyString(), any(FailureType.class), any(WorkerProcess.class), any(String.class), any(String.class)))
                 .thenReturn(false);
 
@@ -140,12 +139,12 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureTypeAtLeastOnce(failureSender, WORKER_EXIT);
+        assertFailureTypeAtLeastOnce(failureHandler, WORKER_EXIT);
     }
 
     @Test
     public void testRun_shouldContinueAfterSendFailure() {
-        when(failureSender.sendFailureOperation(
+        when(failureHandler.handle(
                 anyString(), any(FailureType.class), any(WorkerProcess.class), any(String.class), any(String.class)))
                 .thenReturn(false)
                 .thenReturn(true);
@@ -154,7 +153,7 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureTypeAtLeastOnce(failureSender, WORKER_EXIT);
+        assertFailureTypeAtLeastOnce(failureHandler, WORKER_EXIT);
     }
 
     @Test
@@ -168,7 +167,7 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureTypeAtLeastOnce(failureSender, WORKER_EXCEPTION);
+        assertFailureTypeAtLeastOnce(failureHandler, WORKER_EXCEPTION);
         assertThatExceptionFileDoesNotExist(exceptionFile);
     }
 
@@ -181,7 +180,7 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureTypeAtLeastOnce(failureSender, WORKER_EXCEPTION);
+        assertFailureTypeAtLeastOnce(failureHandler, WORKER_EXCEPTION);
         assertThatExceptionFileDoesNotExist(exceptionFile);
     }
 
@@ -194,13 +193,13 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureTypeAtLeastOnce(failureSender, WORKER_EXCEPTION);
+        assertFailureTypeAtLeastOnce(failureHandler, WORKER_EXCEPTION);
         assertThatExceptionFileDoesNotExist(exceptionFile);
     }
 
     @Test
     public void testRun_shouldDetectException_shouldRenameFileIfFailureOperationCouldNotBeSent_withSingleErrorResponse() {
-        when(failureSender.sendFailureOperation(
+        when(failureHandler.handle(
                 anyString(), any(FailureType.class), any(WorkerProcess.class), any(String.class), any(String.class)))
                 .thenReturn(false)
                 .thenReturn(true);
@@ -212,14 +211,14 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureTypeAtLeastOnce(failureSender, WORKER_EXCEPTION);
+        assertFailureTypeAtLeastOnce(failureHandler, WORKER_EXCEPTION);
         assertThatExceptionFileDoesNotExist(exceptionFile);
         assertThatRenamedExceptionFileExists(exceptionFile);
     }
 
     @Test
     public void testRun_shouldDetectException_shouldRenameFileIfFailureOperationCouldNotBeSent_withContinuousErrorResponse() {
-        when(failureSender.sendFailureOperation(
+        when(failureHandler.handle(
                 anyString(), any(FailureType.class), any(WorkerProcess.class), any(String.class), any(String.class)))
                 .thenReturn(false);
 
@@ -229,7 +228,7 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureTypeAtLeastOnce(failureSender, WORKER_EXCEPTION);
+        assertFailureTypeAtLeastOnce(failureHandler, WORKER_EXCEPTION);
         assertThatExceptionFileDoesNotExist(exceptionFile);
         assertThatRenamedExceptionFileExists(exceptionFile);
     }
@@ -242,7 +241,7 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureType(failureSender, WORKER_OOM);
+        assertFailureType(failureHandler, WORKER_OOM);
     }
 
     @Test
@@ -253,7 +252,7 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureType(failureSender, WORKER_OOM);
+        assertFailureType(failureHandler, WORKER_OOM);
     }
 
     @Test
@@ -265,14 +264,14 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        assertFailureTypeAtLeastOnce(failureSender, WORKER_TIMEOUT);
+        assertFailureTypeAtLeastOnce(failureHandler, WORKER_TIMEOUT);
     }
 
     @Test
     public void testRun_shouldNotDetectInactivity_ifDetectionDisabled() {
         WorkerProcess workerProcess = addRunningWorkerProcess();
 
-        workerProcessFailureMonitor = new WorkerProcessFailureMonitor(failureSender, workerProcessManager, -1,
+        workerProcessFailureMonitor = new WorkerProcessFailureMonitor(failureHandler, workerProcessManager, -1,
                 DEFAULT_CHECK_INTERVAL);
         workerProcessFailureMonitor.start();
 
@@ -286,7 +285,7 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        verifyZeroInteractions(failureSender);
+        verifyZeroInteractions(failureHandler);
     }
 
     @Test
@@ -297,7 +296,7 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        verifyZeroInteractions(failureSender);
+        verifyZeroInteractions(failureHandler);
     }
 
     @Test
@@ -313,7 +312,7 @@ public class WorkerProcessFailureMonitorTest {
 
         sleepMillis(DEFAULT_SLEEP_TIME);
 
-        verifyZeroInteractions(failureSender);
+        verifyZeroInteractions(failureHandler);
     }
 
     @Test(timeout = DEFAULT_TIMEOUT)
@@ -325,7 +324,7 @@ public class WorkerProcessFailureMonitorTest {
         } while (!workerProcess.isFinished());
 
         assertTrue(workerProcess.isFinished());
-        assertFailureType(failureSender, WORKER_FINISHED);
+        assertFailureType(failureHandler, WORKER_FINISHED);
     }
 
     @Test
@@ -335,7 +334,7 @@ public class WorkerProcessFailureMonitorTest {
         sleepMillis(DEFAULT_SLEEP_TIME);
 
         assertFalse(workerProcess.isFinished());
-        assertFailureType(failureSender, WORKER_EXIT);
+        assertFailureType(failureHandler, WORKER_EXIT);
     }
 
     @Test
@@ -395,21 +394,21 @@ public class WorkerProcessFailureMonitorTest {
         return exceptionFile;
     }
 
-    private void assertFailureType(FailureSender failureSender, FailureType failureType) {
-        assertFailureTypeAtLeastOnce(failureSender, failureType, times(1));
+    private void assertFailureType(FailureHandler failureHandler, FailureType failureType) {
+        assertFailureTypeAtLeastOnce(failureHandler, failureType, times(1));
     }
 
-    private void assertFailureTypeAtLeastOnce(FailureSender failureSender, FailureType failureType) {
-        assertFailureTypeAtLeastOnce(failureSender, failureType, atLeastOnce());
+    private void assertFailureTypeAtLeastOnce(FailureHandler failureHandler, FailureType failureType) {
+        assertFailureTypeAtLeastOnce(failureHandler, failureType, atLeastOnce());
     }
 
-    private void assertFailureTypeAtLeastOnce(FailureSender failureSender, FailureType failureType, VerificationMode mode) {
-        verify(failureSender, mode).sendFailureOperation(anyString(),
+    private void assertFailureTypeAtLeastOnce(FailureHandler failureHandler, FailureType failureType, VerificationMode mode) {
+        verify(failureHandler, mode).handle(anyString(),
                 eq(failureType),
                 any(WorkerProcess.class),
                 any(String.class),
                 any(String.class));
-        verifyNoMoreInteractions(failureSender);
+        verifyNoMoreInteractions(failureHandler);
     }
 
     private static void assertThatExceptionFileDoesNotExist(final File firstExceptionFile) {