[SPARK-4991][CORE] Worker should reconnect to Master when Master actor restart

core/src/main/scala/org/apache/spark/deploy/DeployMessage.scala

@@ -89,6 +89,8 @@ private[deploy] object DeployMessages {
   case class KillDriver(driverId: String) extends DeployMessage
 
   case class ApplicationFinished(id: String)
+
+  case class MasterDisconnected(masterUrl: String) extends DeployMessage
 
   // Worker internal
 

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -366,8 +366,12 @@ private[spark] class Master(
               " Asking it to re-register.")
             sender ! ReconnectWorker(masterUrl)
           } else {
+            // Get unknown worker's heart beat, tell the worker disconnected. And worker need to
+            // register to this master first.
             logWarning(s"Got heartbeat from unregistered worker $workerId." +
-              " This worker was never registered, so ignoring the heartbeat.")
+              " This worker was never registered, tell the worker connection is disconnected." +
+              " Need to re-register if want to connect.")
+            sender ! MasterDisconnected(masterUrl)
           }
       }
     }

core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala

@@ -464,6 +464,14 @@ private[spark] class Worker(
     case ApplicationFinished(id) =>
       finishedApps += id
       maybeCleanupApplication(id)
+
+    case MasterDisconnected(masterUrl) =>
+      if (masterUrl != activeMasterUrl) {
+        logWarning(s"Get message from Invalid Master ($masterUrl)." +
+          s"Valid Master is : $activeMasterUrl, so ignore the message.")
+      } else {
+        masterDisconnected()
+      }
   }
 
   private def masterDisconnected() {