diff --git a/mysos/executor/executor.py b/mysos/executor/executor.py
index 122bc36..c3d17ff 100644
--- a/mysos/executor/executor.py
+++ b/mysos/executor/executor.py
@@ -1,4 +1,6 @@
 import json
+import sys
+from threading import Event
 import traceback
 
 from mysos.common.decorators import logged
@@ -9,6 +11,7 @@
 import mesos.interface.mesos_pb2 as mesos_pb2
 from twitter.common import log
 from twitter.common.concurrent import defer
+from twitter.common.quantity import Amount, Time
 
 
 class MysosExecutor(Executor):
@@ -16,6 +19,8 @@
     MysosExecutor is a fine-grained executor, i.e., one executor executes a single task.
   """
 
+  STOP_WAIT = Amount(5, Time.SECONDS)
+
   def __init__(self, runner_provider, sandbox):
     """
       :param runner_provider: An implementation of TaskRunnerProvider.
@@ -27,6 +32,8 @@
     self._killed = False  # True if the executor's singleton task is killed by the scheduler.
     self._sandbox = sandbox
 
+    self._terminated = Event()  # Set when the runner has terminated.
+
   # --- Mesos methods. ---
   @logged
   def registered(self, driver, executorInfo, frameworkInfo, slaveInfo):
@@ -81,21 +88,24 @@
       log.error(traceback.format_exc())
       # Send TASK_LOST for unknown errors.
       self._send_update(task.task_id.value, mesos_pb2.TASK_LOST)
-
-    # Wait for the task's return code (when it terminates).
-    try:
-      returncode = self._runner.join()
-      # Regardless of the return code, if '_runner' terminates, it failed!
-      log.error("Task process terminated with return code %s" % returncode)
-    except TaskError as e:
-      log.error("Task terminated: %s" % e)
-
-    if self._killed:
-      self._send_update(task.task_id.value, mesos_pb2.TASK_KILLED)
     else:
-      self._send_update(task.task_id.value, mesos_pb2.TASK_FAILED)
-
-    self._kill()
+      # Wait for the task's return code (when it terminates).
+      try:
+        returncode = self._runner.join()
+        # If '_runner' terminates, it has either failed or been killed.
+        log.warn("Task process terminated with return code %s" % returncode)
+      except TaskError as e:
+        log.error("Task terminated: %s" % e)
+      finally:
+        if self._killed:
+          self._send_update(task.task_id.value, mesos_pb2.TASK_KILLED)
+        else:
+          self._send_update(task.task_id.value, mesos_pb2.TASK_FAILED)
+        self._terminated.set()
+    finally:
+      # No matter what happens above, when we reach here the executor has no task to run so it
+      # should just commit seppuku.
+      self._kill()
 
   @logged
   def frameworkMessage(self, driver, message):
@@ -118,37 +128,37 @@
           'epoch': master_epoch,  # Send the epoch back without parsing it.
           'position': position
       }))
-    except TaskError as e:
-      # Log the error and do not reply to the framework.
+    except Exception as e:
       log.error("Committing suicide due to failure to process framework message: %s" % e)
+      log.error(traceback.format_exc())
       self._kill()
 
   @logged
   def killTask(self, driver, taskId):
     # Killing the task also kills the executor because there is one task per executor.
     log.info("Asked to kill task %s" % taskId.value)
-    self._killed = True
+
     self._kill()
 
   def _kill(self):
     if self._runner:
+      self._killed = True
       self._runner.stop()  # It could be already stopped. If so, self._runner.stop() is a no-op.
+      self._terminated.wait(sys.maxint)
 
     assert self._driver
 
     # TODO(jyx): Fix https://issues.apache.org/jira/browse/MESOS-243.
-    self._driver.stop()
+    defer(lambda: self._driver.stop(), delay=self.STOP_WAIT)
 
   @logged
   def shutdown(self, driver):
     log.info("Asked to shut down")
-    self._killed = True
     self._kill()
 
   @logged
   def error(self, driver, message):
     log.error("Shutting down due to error: %s" % message)
-    self._killed = True
     self._kill()
 
   def _send_update(self, task_id, state, message=None):
diff --git a/mysos/executor/mysos_task_runner.py b/mysos/executor/mysos_task_runner.py
index 0051bcb..eb75076 100644
--- a/mysos/executor/mysos_task_runner.py
+++ b/mysos/executor/mysos_task_runner.py
@@ -148,6 +148,15 @@
     self._exited.set()
 
   def stop(self, timeout=10):
+    with self._lock:
+      # stop() could be called by multiple threads. Locking so we only stop the runner once.
+      if self._stopping:
+        log.warn("The runner is already stopping/stopped")
+        return False
+      else:
+        log.info("Stopping runner")
+        self._stopping = True
+
     try:
       return self._stop(timeout)
     finally:
@@ -163,17 +172,10 @@
       :return: True if an active runner is stopped, False if the runner is not started or already
                stopping/stopped.
     """
-    if not self._started:
-      log.warn("Cannot stop the runner because it's not started")
-      return False
-
-    if self._stopping:
-      log.warn("The runner is already stopping/stopped")
-      return False
-
     with self._lock:
-      log.info("Stopping runner")
-      self._stopping = True
+      if not self._started:
+        log.warn("Cannot stop the runner because it's not started")
+        return False
 
       if not self._popen:
         log.info("The runner task did not start successfully so no need to kill it")
@@ -198,6 +200,8 @@
         except OSError as e:
           log.info("The sub-processes are already terminated: %s" % e)
           return False
+    else:
+      return True
 
     log.info("Waiting for process to terminate due to SIGKILL")
     if not self._exited.wait(timeout=timeout):
diff --git a/mysos/executor/mysql_task_control.py b/mysos/executor/mysql_task_control.py
index d5ca96f..40504c9 100644
--- a/mysos/executor/mysql_task_control.py
+++ b/mysos/executor/mysql_task_control.py
@@ -150,7 +150,7 @@
             conf_file=self._conf_file,
             buffer_pool_size=self._buffer_pool_size))
     log.info("Executing command: %s" % command)
-    self._process = subprocess.Popen(command, shell=True, env=env)
+    self._process = subprocess.Popen(command, shell=True, env=env, preexec_fn=os.setpgrp)
 
     # There is a delay before mysqld becomes available to accept requests. Wait for it.
     command = "%(cmd)s %(pid_file)s %(port)s %(timeout)s" % dict(
diff --git a/tests/scheduler/test_mysos_scheduler.py b/tests/scheduler/test_mysos_scheduler.py
index 3aeead1..ffb318b 100644
--- a/tests/scheduler/test_mysos_scheduler.py
+++ b/tests/scheduler/test_mysos_scheduler.py
@@ -12,6 +12,7 @@
 from twitter.common import log
 from twitter.common.concurrent import deadline
 from twitter.common.dirutil import safe_mkdtemp
+from twitter.common.metrics import RootMetrics
 from twitter.common.quantity import Amount, Time
 from zake.fake_client import FakeClient
 from zake.fake_storage import FakeStorage
@@ -64,6 +65,8 @@
       "/fakepath",
       gen_encryption_key())
 
+  RootMetrics().register_observable('scheduler', scheduler)
+
   scheduler_driver = mesos.native.MesosSchedulerDriver(
       scheduler,
       framework_info,
@@ -84,11 +87,14 @@
 
   scheduler.delete_cluster(cluster_name, password="passwd")
 
-  # A slave is promoted to be the master.
+  # The cluster is deleted from ZooKeeper.
   deadline(
       lambda: wait_for_termination(
           get_cluster_path(posixpath.join(zk_url, 'discover'), cluster_name),
           zk_client),
       Amount(40, Time.SECONDS))
 
+  sample = RootMetrics().sample()
+  assert sample['scheduler.tasks_killed'] == 1
+
   assert scheduler_driver.stop() == DRIVER_STOPPED