diff --git a/akka-cluster/src/multi-jvm/scala/akka/cluster/ClusterDeathWatchSpec.scala b/akka-cluster/src/multi-jvm/scala/akka/cluster/ClusterDeathWatchSpec.scala
index 467b2e49c0..d6287ed099 100644
--- a/akka-cluster/src/multi-jvm/scala/akka/cluster/ClusterDeathWatchSpec.scala
+++ b/akka-cluster/src/multi-jvm/scala/akka/cluster/ClusterDeathWatchSpec.scala
@@ -10,13 +10,16 @@ import akka.remote.testkit.MultiNodeSpec
 import akka.testkit._
 import akka.actor.Props
 import akka.actor.Actor
+import akka.actor.Address
 import akka.actor.RootActorPath
 import akka.actor.Terminated
+import akka.actor.Address
 
 object ClusterDeathWatchMultiJvmSpec extends MultiNodeConfig {
   val first = role("first")
   val second = role("second")
   val third = role("third")
+  val fourth = role("fourth")
 
   commonConfig(debugConfig(on = false).withFallback(MultiNodeClusterSpec.clusterConfig))
 }
@@ -24,6 +27,7 @@ object ClusterDeathWatchMultiJvmSpec extends MultiNodeConfig {
 class ClusterDeathWatchMultiJvmNode1 extends ClusterDeathWatchSpec with FailureDetectorPuppetStrategy
 class ClusterDeathWatchMultiJvmNode2 extends ClusterDeathWatchSpec with FailureDetectorPuppetStrategy
 class ClusterDeathWatchMultiJvmNode3 extends ClusterDeathWatchSpec with FailureDetectorPuppetStrategy
+class ClusterDeathWatchMultiJvmNode4 extends ClusterDeathWatchSpec with FailureDetectorPuppetStrategy
 
 abstract class ClusterDeathWatchSpec
   extends MultiNodeSpec(ClusterDeathWatchMultiJvmSpec)
@@ -43,19 +47,18 @@ abstract class ClusterDeathWatchSpec
         val path3 = RootActorPath(third) / "user" / "subject"
         val watchEstablished = TestLatch(1)
         system.actorOf(Props(new Actor {
-
           context.watch(context.actorFor(path2))
           context.watch(context.actorFor(path3))
           watchEstablished.countDown
-
           def receive = {
             case t: Terminated ⇒ testActor ! t.actor.path
           }
-        }), name = "observer")
+        }), name = "observer1")
 
         watchEstablished.await
         enterBarrier("watch-established")
         expectMsg(path2)
+        expectNoMsg
         enterBarrier("second-terminated")
 
         markNodeAsUnavailable(third)
@@ -64,7 +67,7 @@ abstract class ClusterDeathWatchSpec
 
       }
 
-      runOn(second, third) {
+      runOn(second, third, fourth) {
         system.actorOf(Props(new Actor { def receive = Actor.emptyBehavior }), name = "subject")
         enterBarrier("subjected-started")
         enterBarrier("watch-established")
@@ -79,5 +82,37 @@ abstract class ClusterDeathWatchSpec
 
     }
 
+    "receive Terminated when watched node is unknown host" taggedAs LongRunningTest in {
+      runOn(first) {
+        val path = RootActorPath(Address("akka", system.name, "unknownhost", 2552)) / "user" / "subject"
+        system.actorOf(Props(new Actor {
+          context.watch(context.actorFor(path))
+          def receive = {
+            case t: Terminated ⇒ testActor ! t.actor.path
+          }
+        }), name = "observer2")
+
+        expectMsg(path)
+      }
+
+      enterBarrier("after")
+    }
+
+    "receive Terminated when watched path doesn't exist" taggedAs LongRunningTest in {
+      runOn(first) {
+        val path = RootActorPath(second) / "user" / "non-existing"
+        system.actorOf(Props(new Actor {
+          context.watch(context.actorFor(path))
+          def receive = {
+            case t: Terminated ⇒ testActor ! t.actor.path
+          }
+        }), name = "observer3")
+
+        expectMsg(path)
+      }
+
+      enterBarrier("after")
+    }
+
   }
 }
diff --git a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala b/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
index 5377013a42..24b38ed868 100644
--- a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
+++ b/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
@@ -236,6 +236,7 @@ private[akka] class RemoteActorRef private[akka] (
     catch {
       case e @ (_: InterruptedException | NonFatal(_)) ⇒
         remote.system.eventStream.publish(Error(e, path.toString, classOf[RemoteActorRef], "swallowing exception during message send"))
+        provider.deadLetters ! message
     }
 
   override def !(message: Any)(implicit sender: ActorRef = null): Unit =
@@ -243,6 +244,7 @@ private[akka] class RemoteActorRef private[akka] (
     catch {
       case e @ (_: InterruptedException | NonFatal(_)) ⇒
         remote.system.eventStream.publish(Error(e, path.toString, classOf[RemoteActorRef], "swallowing exception during message send"))
+        provider.deadLetters ! message
     }
 
   def suspend(): Unit = sendSystemMessage(Suspend())