akka#21574 Fix dispatcher use after system termination

akka-actor-tests/src/test/scala/akka/dispatch/DispatcherShutdownSpec.scala

@@ -0,0 +1,34 @@
+package akka.dispatch
+
+import akka.actor.ActorSystem
+import akka.testkit.TestKit
+import java.lang.management.ManagementFactory
+import org.scalatest.{ Matchers, WordSpec }
+import scala.concurrent.{ Await, Future }
+import scala.concurrent.duration._
+
+class DispatcherShutdownSpec extends WordSpec with Matchers {
+
+  "akka dispatcher" should {
+
+    "eventualy shutdown when used after system terminate" in {
+
+      val threads = ManagementFactory.getThreadMXBean()
+      def threadCount = threads
+        .dumpAllThreads(false, false).toList
+        .map(_.getThreadName)
+        .filter(_.startsWith("DispatcherShutdownSpec-akka.actor.default"))
+        .size
+
+      val system = ActorSystem("DispatcherShutdownSpec")
+      threadCount should be > 0
+
+      Await.ready(system.terminate(), 1.second)
+      Await.ready(Future(akka.Done)(system.dispatcher), 1.second)
+
+      TestKit.awaitCond(threadCount == 0, 3.second)
+    }
+
+  }
+
+}

akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala

@@ -174,7 +174,13 @@ abstract class MessageDispatcher(val configurator: MessageDispatcherConfigurator
       override def execute(runnable: Runnable): Unit = runnable.run()
       override def reportFailure(t: Throwable): Unit = MessageDispatcher.this.reportFailure(t)
     }) catch {
-      case _: IllegalStateException ⇒ shutdown()
+      case _: IllegalStateException ⇒
+        shutdown()
+        // Since there is no scheduler anymore, restore the state to UNSCHEDULED.
+        // When this dispatcher is used again,
+        // shutdown is only attempted if the state is UNSCHEDULED
+        // (as per ifSensibleToDoSoThenScheduleShutdown above)
+        updateShutdownSchedule(SCHEDULED, UNSCHEDULED)
     }
   }