diff --git a/mysos/scheduler/scheduler.py b/mysos/scheduler/scheduler.py
index 0c060df..240d481 100644
--- a/mysos/scheduler/scheduler.py
+++ b/mysos/scheduler/scheduler.py
@@ -238,6 +238,11 @@
       self._total_requested_disk_mb.write(
           self._total_requested_disk_mb.read() - cluster_info.total_disk_mb)
 
+      if launcher.terminated:
+        log.info("Deleting the launcher for cluster %s directly because the cluster has already "
+                 "terminated" % launcher.cluster_name)
+        self._delete_launcher(launcher)
+
       return get_cluster_path(self._discover_zk_url, cluster_name)
 
   @property
@@ -407,9 +412,13 @@
       if launcher.terminated:
         log.info("Deleting the launcher for cluster %s because the cluster has terminated" %
                  launcher.cluster_name)
-        self._state.clusters.discard(launcher.cluster_name)
-        self._state_provider.dump_scheduler_state(self._state)
-        del self._launchers[launcher.cluster_name]
+        self._delete_launcher(launcher)
+
+  def _delete_launcher(self, launcher):
+    assert launcher.terminated
+    self._state.clusters.discard(launcher.cluster_name)
+    self._state_provider.dump_scheduler_state(self._state)
+    del self._launchers[launcher.cluster_name]
 
   @logged
   def frameworkMessage(self, driver, executorId, slaveId, message):
diff --git a/tests/scheduler/test_scheduler.py b/tests/scheduler/test_scheduler.py
index 58ea8a3..deca345 100644
--- a/tests/scheduler/test_scheduler.py
+++ b/tests/scheduler/test_scheduler.py
@@ -209,16 +209,16 @@
     scheduler_key = gen_encryption_key()
 
     scheduler = MysosScheduler(
-      self._state,
-      self._state_provider,
-      self._framework_user,
-      "./executor.pex",
-      "cmd.sh",
-      self._zk_client,
-      self._zk_url,
-      Amount(5, Time.SECONDS),
-      "/etc/mysos/admin_keyfile.yml",
-      scheduler_key)
+        self._state,
+        self._state_provider,
+        self._framework_user,
+        "./executor.pex",
+        "cmd.sh",
+        self._zk_client,
+        self._zk_url,
+        Amount(5, Time.SECONDS),
+        "/etc/mysos/admin_keyfile.yml",
+        scheduler_key)
 
     RootMetrics().register_observable('scheduler', scheduler)
 
@@ -238,3 +238,28 @@
     assert sample['scheduler.total_requested_mem_mb'] == 0
     assert sample['scheduler.total_requested_disk_mb'] == 0
     assert sample['scheduler.total_requested_cpus'] == 0
+
+  def test_scheduler_delete_empty_cluster(self):
+    scheduler_key = gen_encryption_key()
+
+    scheduler = MysosScheduler(
+        self._state,
+        self._state_provider,
+        self._framework_user,
+        "./executor.pex",
+        "cmd.sh",
+        self._zk_client,
+        self._zk_url,
+        Amount(5, Time.SECONDS),
+        "/etc/mysos/admin_keyfile.yml",
+        scheduler_key)
+
+    scheduler.registered(self._driver, self._framework_id, object())
+    _, password = scheduler.create_cluster("cluster1", "mysql_user", 3)
+
+    assert len(scheduler._launchers) == 1
+
+    # Deleting the cluster before any offer comes in for launching any task.
+    scheduler.delete_cluster("cluster1", password)
+
+    assert len(scheduler._launchers) == 0