diff --git a/core/src/main/scala/kafka/controller/ControllerEventManager.scala b/core/src/main/scala/kafka/controller/ControllerEventManager.scala
index 54e3a9e126b6c..a456ce32895a9 100644
--- a/core/src/main/scala/kafka/controller/ControllerEventManager.scala
+++ b/core/src/main/scala/kafka/controller/ControllerEventManager.scala
@@ -61,6 +61,7 @@ class ControllerEventManager(controllerId: Int, rateAndTimeMetrics: Map[Controll
   def start(): Unit = thread.start()
 
   def close(): Unit = {
+    thread.initiateShutdown()
     clearAndPut(KafkaController.ShutdownEventThread)
     thread.awaitShutdown()
   }
@@ -83,7 +84,7 @@ class ControllerEventManager(controllerId: Int, rateAndTimeMetrics: Map[Controll
 
     override def doWork(): Unit = {
       queue.take() match {
-        case KafkaController.ShutdownEventThread => initiateShutdown()
+        case KafkaController.ShutdownEventThread => // The shutting down of the thread has been initiated at this point. Ignore this event.
         case controllerEvent =>
           _state = controllerEvent.state
 
diff --git a/core/src/main/scala/kafka/utils/ShutdownableThread.scala b/core/src/main/scala/kafka/utils/ShutdownableThread.scala
index 13bbc90f2a5c7..02d09dab96d82 100644
--- a/core/src/main/scala/kafka/utils/ShutdownableThread.scala
+++ b/core/src/main/scala/kafka/utils/ShutdownableThread.scala
@@ -27,7 +27,8 @@ abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean
   this.logIdent = "[" + name + "]: "
   private val shutdownInitiated = new CountDownLatch(1)
   private val shutdownComplete = new CountDownLatch(1)
-
+  @volatile private var isStarted: Boolean = false
+  
   def shutdown(): Unit = {
     initiateShutdown()
     awaitShutdown()
@@ -54,8 +55,13 @@ abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean
    * After calling initiateShutdown(), use this API to wait until the shutdown is complete
    */
   def awaitShutdown(): Unit = {
-    shutdownComplete.await()
-    info("Shutdown completed")
+    if (shutdownInitiated.getCount != 0)
+      throw new IllegalStateException("initiateShutdown() was not called before awaitShutdown()")
+    else {
+      if (isStarted)
+        shutdownComplete.await()
+      info("Shutdown completed")
+    }
   }
 
   /**
@@ -76,6 +82,7 @@ abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean
   def doWork(): Unit
 
   override def run(): Unit = {
+    isStarted = true
     info("Starting")
     try {
       while (isRunning)