一Master故障挥着宕机，可能触发新的Master选举

当重新选择Leader的时候，会进行集群的恢复，在恢复的过程中，就会向Worker和AppClient发送MasterChanged消息。

private def beginRecovery(storedApps: Seq[ApplicationInfo], storedDrivers: Seq[DriverInfo],
 storedWorkers: Seq[WorkerInfo]) {
 // 遍历每一个存储的application，注册该application,并且发送MasterChanged请求
 for (app <- storedApps) {
 logInfo("Trying to recover app: " + app.id)
 try {
 registerApplication(app)
 // 将该application状态置为UNKNOWN状态
 app.state = ApplicationState.UNKNOWN
 // 然后这个app向master发送MasterChanged请求
 app.driver.send(MasterChanged(self, masterWebUiUrl))
 } catch {
 case e: Exception => logInfo("App " + app.id + " had exception on reconnect")
 }
 }
 // 遍历每一个存储的driver, 更新master所维护的driver集合
 for (driver <- storedDrivers) {
 drivers += driver
 }
 // 遍历每一个存储的wroker,然后向master注册worker
 for (worker <- storedWorkers) {
 logInfo("Trying to recover worker: " + worker.id)
 try {
 // 注册worker，就是更新master的woker集合，和worker相关的映射列表
 registerWorker(worker)
 // 将该worker状态置为UNKNOWN状态
 worker.state = WorkerState.UNKNOWN
 // 然后改worker向master发送MasterChanged请求
 worker.endpoint.send(MasterChanged(self, masterWebUiUrl))
 } catch {
 case e: Exception => logInfo("Worker " + worker.id + " had exception on reconnect")
 }
 }
}

二Worker和AppClient会接受到来自Master的MasterChanged消息

2.1 Worker在收到MasterChanged消息

# 获取新的master的url和master，连接状态置为true,取消之前的尝试重新注册

# 向新的master发送WorkerSchedulerStateResponse消息，然后会做一些操作

case MasterChanged(masterRef, masterWebUiUrl) =>
logInfo("Master has changed, new master is at " + masterRef.address.toSparkURL)
// 获取新的master的url和master，连接状态置为true,取消之前的尝试重新注册
changeMaster(masterRef, masterWebUiUrl)
// 创建当前节点executors的简单描述对象ExecutorDescription
val execs = executors.values.
map(e => new ExecutorDescription(e.appId, e.execId, e.cores, e.state))
// 向新的master发送WorkerSchedulerStateResponse消息，然后会做一些操作
masterRef.send(WorkerSchedulerStateResponse(workerId, execs.toList, drivers.keys.toSeq))

2.2 AppClient在收到MasterChanged消息

# 更新master

# 向新的master发送MasterChangeAcknowledged消息

case MasterChanged(masterRef, masterWebUiUrl) =>
 logInfo("Master has changed, new master is at " + masterRef.address.toSparkURL)
 // 更新master
 master = Some(masterRef)
 alreadyDisconnected = false
 // 向新的master发送MasterChangeAcknowledged消息
 masterRef.send(MasterChangeAcknowledged(appId.get))

三 Master会接受来自Worker和AppClient的消息

3.1 Master在收到Worker的WorkerSchedulerStateResponse消息

由于这是新的master,所以worker需要重新注册，然后新的master再次把之前相关的应用程序在worker上进行恢复

case WorkerSchedulerStateResponse(workerId, executors, driverIds) =>
 // 根据workerId获取worker
 idToWorker.get(workerId) match {
 case Some(worker) =>
 logInfo("Worker has been re-registered: " + workerId)
 // worker状态置为alive
 worker.state = WorkerState.ALIVE
 // 从指定的executor中过滤出哪些是有效的executor
 val validExecutors = executors.filter(exec => idToApp.get(exec.appId).isDefined)
 // 遍历有效的executors
 for (exec <- validExecutors) {
 // 获取executor所对应的app
 val app = idToApp.get(exec.appId).get
 // 为app设置executor，比如哪一个worker,多少核数等资源
 val execInfo = app.addExecutor(worker, exec.cores, Some(exec.execId))
 // 将该executor添加到该woker上
 worker.addExecutor(execInfo)
 execInfo.copyState(exec)
 }
 // 将所有的driver设置为RUNNING然后加入到worker中
 for (driverId <- driverIds) {
 drivers.find(_.id == driverId).foreach { driver =>
 driver.worker = Some(worker)
 driver.state = DriverState.RUNNING
 worker.drivers(driverId) = driver
 }
 }
 case None =>
 logWarning("Scheduler state from unknown worker: " + workerId)
 }
 // 判断当前是否可以进行completeRecovery操作，如果可以进行completeRecovery操作
 if (canCompleteRecovery) { completeRecovery() }

3.2 Master收到AppClient的MasterChangeAcknowledged消息

# 更新application状态为WAITTING

# 判断当前是否可以进行completeRecovery操作，如果可以进行completeRecovery操作

case MasterChangeAcknowledged(appId) =>
 idToApp.get(appId) match {
 case Some(app) =>
 logInfo("Application has been re-registered: " + appId)
 app.state = ApplicationState.WAITING
 case None =>
 logWarning("Master change ack from unknown app: " + appId)
 }
 // 判断当前是否可以进行completeRecovery操作，如果可以进行completeRecovery操作
 if (canCompleteRecovery) { completeRecovery() }