Send ChildTerminated before Terminated when AddressTerminated, see #2983

* Replaces the previous half-baked removeChildWhenToAddressTerminated

akka-actor/src/main/scala/akka/actor/ActorCell.scala

@@ -408,10 +408,8 @@ private[akka] class ActorCell(
         publish(Debug(self.path.toString, clazz(actor), "received AutoReceiveMessage " + msg))
 
       msg.message match {
-        case Failed(cause, uid) ⇒ handleFailure(sender, cause, uid)
-        case t: Terminated ⇒
-          if (t.addressTerminated) removeChildWhenToAddressTerminated(t.actor)
-          watchedActorTerminated(t)
+        case Failed(cause, uid)         ⇒ handleFailure(sender, cause, uid)
+        case t: Terminated              ⇒ watchedActorTerminated(t)
         case AddressTerminated(address) ⇒ addressTerminated(address)
         case Kill                       ⇒ throw new ActorKilledException("Kill")
         case PoisonPill                 ⇒ self.stop()
@@ -421,18 +419,6 @@ private[akka] class ActorCell(
       }
     }
 
-  /**
-   * When a parent is watching a child and it terminates due to AddressTerminated,
-   * it should be removed to support immediate creation of child with same name.
-   *
-   * For remote deployed actors ChildTerminated should be sent to the supervisor
-   * to clean up child references of remote deployed actors when remote node
-   * goes down, i.e. triggered by AddressTerminated, but that is the responsibility
-   * of the ActorRefProvider to handle that scenario.
-   */
-  private def removeChildWhenToAddressTerminated(child: ActorRef): Unit =
-    childrenRefs.getByRef(child) foreach { crs ⇒ removeChildAndGetStateChange(crs.child) }
-
   final def receiveMessage(msg: Any): Unit = behaviorStack.head.applyOrElse(msg, actor.unhandled)
 
   /*

akka-actor/src/main/scala/akka/actor/dungeon/DeathWatch.scala

@@ -5,7 +5,7 @@
 package akka.actor.dungeon
 
 import akka.actor.{ Terminated, InternalActorRef, ActorRef, ActorRefScope, ActorCell, Actor, Address, AddressTerminated }
-import akka.dispatch.{ Watch, Unwatch }
+import akka.dispatch.{ ChildTerminated, Watch, Unwatch }
 import akka.event.Logging.{ Warning, Error, Debug }
 import scala.util.control.NonFatal
 
@@ -130,7 +130,11 @@ private[akka] trait DeathWatch { this: ActorCell ⇒
     // send Terminated to self for all matching subjects
     // existenceConfirmed = false because we could have been watching a
     // non-local ActorRef that had never resolved before the other node went down
+    // When a parent is watching a child and it terminates due to AddressTerminated
+    // it is removed by sending ChildTerminated to support immediate creation of child
+    // with same name.
     for (a ← watching; if a.path.address == address) {
+      childrenRefs.getByRef(a) foreach { _ ⇒ self.sendSystemMessage(ChildTerminated(a)) }
       self ! Terminated(a)(existenceConfirmed = false, addressTerminated = true)
     }
   }

akka-docs/rst/cluster/cluster-usage-java.rst

@@ -320,6 +320,16 @@ This is how the curve looks like for ``acceptable-heartbeat-pause`` configured t
 
 .. image:: images/phi3.png
 
+Death watch uses the cluster failure detector for nodes in the cluster, i.e. it 
+generates ``Terminated`` message from network failures and JVM crashes, in addition 
+to graceful termination of watched actor. 
+
+.. warning::
+
+  Creating a remote deployed child actor with the same name as the terminated 
+  actor is not fully supported. There is a race condition that potentially removes the new 
+  actor.
+
 Cluster Aware Routers
 ^^^^^^^^^^^^^^^^^^^^^
 

akka-docs/rst/cluster/cluster-usage-scala.rst

@@ -293,6 +293,17 @@ This is how the curve looks like for ``acceptable-heartbeat-pause`` configured t
 
 .. image:: images/phi3.png
 
+
+Death watch uses the cluster failure detector for nodes in the cluster, i.e. it 
+generates ``Terminated`` message from network failures and JVM crashes, in addition 
+to graceful termination of watched actor. 
+
+.. warning::
+
+  Creating a remote deployed child actor with the same name as the terminated 
+  actor is not fully supported. There is a race condition that potentially removes the new 
+  actor.
+
 .. _cluster_aware_routers_scala:
 
 Cluster Aware Routers