diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManager.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManager.java
index 61564b1..c555adc 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManager.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorManager.java
@@ -263,15 +263,18 @@
         try {
           // We need to wait awhile before returning the container to the RM
           // in order to give the EvaluatorRuntime (and Launcher) time to cleanly exit.
-          this.clock.scheduleAlarm(100, new EventHandler<Alarm>() {
+          this.clock.scheduleAlarm(200, new EventHandler<Alarm>() {
             @Override
             public void onNext(final Alarm alarm) {
+              LOG.log(Level.FINER, "Close EvaluatorManager {0} - release to RM", evaluatorId);
               resourceReleaseHandler.onNext(releaseEvent);
+              shutdown();
             }
           });
         } catch (final IllegalStateException e) {
           LOG.log(Level.WARNING, "Force resource release because the client closed the clock.", e);
           this.resourceReleaseHandler.onNext(releaseEvent);
+          this.shutdown();
         }
       }
     }
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorMessageDispatcher.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorMessageDispatcher.java
index 73854b2..e113e76 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorMessageDispatcher.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/driver/evaluator/EvaluatorMessageDispatcher.java
@@ -290,10 +290,10 @@
     LOG.log(Level.FINER, "Closing message dispatcher for {0}", this.evaluatorIdentifier);
     // This effectively closes all dispatchers as they share the same stage.
     this.serviceDispatcher.close();
-    if (!this.serviceDispatcher.isThreadPoolClosed()) {
+    if (!this.serviceDispatcher.isClosed()) {
       LOG.log(Level.SEVERE,
-              "Closing message dispatcher for {0}: ThreadPool for service dispatcher failed to close",
-              this.evaluatorIdentifier);
+          "Closing message dispatcher for {0}: ThreadPool for service dispatcher failed to close",
+          this.evaluatorIdentifier);
     }
   }
 }
diff --git a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/utils/DispatchingEStage.java b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/utils/DispatchingEStage.java
index f5b1c1d..91ff7d6 100644
--- a/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/utils/DispatchingEStage.java
+++ b/lang/java/reef-common/src/main/java/org/apache/reef/runtime/common/utils/DispatchingEStage.java
@@ -28,6 +28,8 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 /**
  * Delayed event router that dispatches messages to the proper event handler by type.
@@ -37,6 +39,8 @@
 @DriverSide
 public final class DispatchingEStage implements AutoCloseable {
 
+  private static final Logger LOG = Logger.getLogger(DispatchingEStage.class.getName());
+
   /**
    * A map of event handlers, populated in the register() method.
    */
@@ -98,7 +102,7 @@
 
   /**
    * Dispatch a new message by type.
-   *
+   * If the stage is already closed, log a warning and ignore the message.
    * @param type    Type of event handler - must match the register() call.
    * @param message A message to process. Must be a subclass of T.
    * @param <T>     Message type that event handler supports.
@@ -106,8 +110,13 @@
    */
   @SuppressWarnings("unchecked")
   public <T, U extends T> void onNext(final Class<T> type, final U message) {
-    final EventHandler<T> handler = (EventHandler<T>) this.handlers.get(type);
-    this.stage.onNext(new DelayedOnNext(handler, message));
+    if (this.isClosed()) {
+      LOG.log(Level.WARNING, "Dispatcher {0} already closed: ignoring message {1}: {2}",
+          new Object[] {this.stage, type.getCanonicalName(), message});
+    } else {
+      final EventHandler<T> handler = (EventHandler<T>) this.handlers.get(type);
+      this.stage.onNext(new DelayedOnNext(handler, message));
+    }
   }
 
   /**
@@ -118,7 +127,8 @@
   }
 
   /**
-   * Close the internal thread pool.
+   * Close the stage adn stop accepting new messages.
+   * Closes the internal thread pool.
    */
   @Override
   public void close() {
@@ -126,9 +136,10 @@
   }
 
   /**
-   * Returns true if the internal thread pool is closed.
+   * Check if the stage can still accept messages.
+   * @return true if the stage can no longer accept messages, false otherwise.
    */
-  public boolean isThreadPoolClosed() {
+  public boolean isClosed() {
     return this.stage.isClosed();
   }
 
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/AbstractEStage.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/AbstractEStage.java
index 2861204..5395054 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/AbstractEStage.java
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/AbstractEStage.java
@@ -88,4 +88,20 @@
     outMeter.mark(1);
   }
 
+  /**
+   * Check if the stage can still accept messages.
+   * @return true if the stage is closed, false otherwise.
+   */
+  public boolean isClosed() {
+    return closed.get();
+  }
+
+  /**
+   * Get human readable representation of the class (used for logging).
+   * @return A string that contains stage name.
+   */
+  @Override
+  public String toString() {
+    return String.format("Stage:%s:%s", this.getClass().getCanonicalName(), name);
+  }
 }
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/impl/ThreadPoolStage.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/impl/ThreadPoolStage.java
index 005db44..4c57fa7 100644
--- a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/impl/ThreadPoolStage.java
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/impl/ThreadPoolStage.java
@@ -236,13 +236,6 @@
   }
 
   /**
-   * Returns true if resources are closed.
-   */
-  public boolean isClosed() {
-    return closed.get() && executor.isTerminated();
-  }
-
-  /**
    * Gets the queue length of this stage.
    *
    * @return the queue length