Flink源码系列(TaskExecutor向ResourceManager发起注册[flink内部，非yarn中rm])-第十期

上一期指路：

上一期​​​​​​​

承接上一期讲到YarnTaskExecutorRunner的main方法，我们继续往下分析。

1.YarnTaskExecutorRunner#main->YarnTaskExecutorRunner#runTaskManagerSecurely->TaskManagerRunner#runTaskManagerSecurely

	public static void runTaskManagerSecurely(Configuration configuration) throws Exception {replaceGracefulExitWithHaltIfConfigured(configuration);final PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(configuration);FileSystem.initialize(configuration, pluginManager);SecurityUtils.install(new SecurityConfiguration(configuration));SecurityUtils.getInstalledContext().runSecured(() -> {runTaskManager(configuration, pluginManager);return null;});}

2. TaskManagerRunner#runTaskManager

	public static void runTaskManager(Configuration configuration, PluginManager pluginManager) throws Exception {final TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager, TaskManagerRunner::createTaskExecutorService);taskManagerRunner.start();}

①new TaskManagerRunner

构建TaskManagerRunner

②taskManagerRunner.start()

启动

3.TaskManagerRunner#start->TaskExecutorToServiceAdapter#start->RpcEndpoint#start

rpcServer.start()

rpc服务启动。即发消息通知底层的 AkkaRpcActor 切换为 START 状态。那么直接看TaskExecutor的onStart方法

4.TaskExecutor#onStart->TaskExecutor#startTaskExecutorServices

	private void startTaskExecutorServices() throws Exception {try {// start by connecting to the ResourceManagerresourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());// tell the task slot table who's responsible for the task slot actionstaskSlotTable.start(new SlotActionsImpl(), getMainThreadExecutor());// start the job leader servicejobLeaderService.start(getAddress(), getRpcService(), haServices, new JobLeaderListenerImpl());fileCache = new FileCache(taskManagerConfiguration.getTmpDirectories(), blobCacheService.getPermanentBlobService());} catch (Exception e) {handleStartTaskExecutorServicesException(e);}}

5.StandaloneLeaderRetrievalService#start->TaskExecutor的内部类ResourceManagerLeaderListener#notifyLeaderAddress->TaskExecutor#notifyOfNewResourceManagerLeader->TaskExecutor#reconnectToResourceManager->TaskExecutor#tryConnectToResourceManager->TaskExecutor#connectToResourceManager

	private void connectToResourceManager() {assert(resourceManagerAddress != null);assert(establishedResourceManagerConnection == null);assert(resourceManagerConnection == null);log.info("Connecting to ResourceManager {}.", resourceManagerAddress);final TaskExecutorRegistration taskExecutorRegistration = new TaskExecutorRegistration(getAddress(),getResourceID(),unresolvedTaskManagerLocation.getDataPort(),JMXService.getPort().orElse(-1),hardwareDescription,memoryConfiguration,taskManagerConfiguration.getDefaultSlotResourceProfile(),taskManagerConfiguration.getTotalResourceProfile());resourceManagerConnection =new TaskExecutorToResourceManagerConnection(log,getRpcService(),taskManagerConfiguration.getRetryingRegistrationConfiguration(),resourceManagerAddress.getAddress(),resourceManagerAddress.getResourceManagerId(),getMainThreadExecutor(),new ResourceManagerRegistrationListener(),taskExecutorRegistration);resourceManagerConnection.start();}

6.RegisteredRpcConnection#start

	public void start() {checkState(!closed, "The RPC connection is already closed");checkState(!isConnected() && pendingRegistration == null, "The RPC connection is already started");final RetryingRegistration<F, G, S> newRegistration = createNewRegistration();if (REGISTRATION_UPDATER.compareAndSet(this, null, newRegistration)) {newRegistration.startRegistration();} else {// concurrent start operationnewRegistration.cancel();}}

①createNewRegistration

创建一个te向rm发起注册

②startRegistration

启动这个注册

7.RetryingRegistration#startRegistration

	public void startRegistration() {if (canceled) {// we already got canceledreturn;}try {// trigger resolution of the target address to a callable gatewayfinal CompletableFuture<G> rpcGatewayFuture;if (FencedRpcGateway.class.isAssignableFrom(targetType)) {rpcGatewayFuture = (CompletableFuture<G>) rpcService.connect(targetAddress,fencingToken,targetType.asSubclass(FencedRpcGateway.class));} else {rpcGatewayFuture = rpcService.connect(targetAddress, targetType);}// upon success, start the registration attemptsCompletableFuture<Void> rpcGatewayAcceptFuture = rpcGatewayFuture.thenAcceptAsync((G rpcGateway) -> {log.info("Resolved {} address, beginning registration", targetName);register(rpcGateway, 1, retryingRegistrationConfiguration.getInitialRegistrationTimeoutMillis());},rpcService.getExecutor());// upon failure, retry, unless this is cancelledrpcGatewayAcceptFuture.whenCompleteAsync((Void v, Throwable failure) -> {if (failure != null && !canceled) {final Throwable strippedFailure = ExceptionUtils.stripCompletionException(failure);if (log.isDebugEnabled()) {log.debug("Could not resolve {} address {}, retrying in {} ms.",targetName,targetAddress,retryingRegistrationConfiguration.getErrorDelayMillis(),strippedFailure);} else {log.info("Could not resolve {} address {}, retrying in {} ms: {}",targetName,targetAddress,retryingRegistrationConfiguration.getErrorDelayMillis(),strippedFailure.getMessage());}startRegistrationLater(retryingRegistrationConfiguration.getErrorDelayMillis());}},rpcService.getExecutor());}catch (Throwable t) {completionFuture.completeExceptionally(t);cancel();}}

①rpcService.connect

将目标地址解析为一个可调用的网关

②register

成功后，就开始尝试注册

8.RetryingRegistration#register->TaskExecutorToResourceManagerConnection的内部类ResourceManagerRegistration#invokeRegistration->ResourceManager#registerTaskExecutorInternal

	private RegistrationResponse registerTaskExecutorInternal(TaskExecutorGateway taskExecutorGateway,TaskExecutorRegistration taskExecutorRegistration) {ResourceID taskExecutorResourceId = taskExecutorRegistration.getResourceId();WorkerRegistration<WorkerType> oldRegistration = taskExecutors.remove(taskExecutorResourceId);if (oldRegistration != null) {// TODO :: suggest old taskExecutor to stop itselflog.debug("Replacing old registration of TaskExecutor {}.", taskExecutorResourceId.getStringWithMetadata());// remove old task manager registration from slot managerslotManager.unregisterTaskManager(oldRegistration.getInstanceID(),new ResourceManagerException(String.format("TaskExecutor %s re-connected to the ResourceManager.", taskExecutorResourceId.getStringWithMetadata())));}final WorkerType newWorker = workerStarted(taskExecutorResourceId);String taskExecutorAddress = taskExecutorRegistration.getTaskExecutorAddress();if (newWorker == null) {log.warn("Discard registration from TaskExecutor {} at ({}) because the framework did " +"not recognize it", taskExecutorResourceId.getStringWithMetadata(), taskExecutorAddress);return new RegistrationResponse.Decline("unrecognized TaskExecutor");} else {WorkerRegistration<WorkerType> registration = new WorkerRegistration<>(taskExecutorGateway,newWorker,taskExecutorRegistration.getDataPort(),taskExecutorRegistration.getJmxPort(),taskExecutorRegistration.getHardwareDescription(),taskExecutorRegistration.getMemoryConfiguration());log.info("Registering TaskManager with ResourceID {} ({}) at ResourceManager", taskExecutorResourceId.getStringWithMetadata(), taskExecutorAddress);taskExecutors.put(taskExecutorResourceId, registration);taskManagerHeartbeatManager.monitorTarget(taskExecutorResourceId, new HeartbeatTarget<Void>() {@Overridepublic void receiveHeartbeat(ResourceID resourceID, Void payload) {// the ResourceManager will always send heartbeat requests to the// TaskManager}@Overridepublic void requestHeartbeat(ResourceID resourceID, Void payload) {taskExecutorGateway.heartbeatFromResourceManager(resourceID);}});return new TaskExecutorRegistrationSuccess(registration.getInstanceID(),resourceId,clusterInformation);}}

①getResourceId

获取te的资源id

②taskExecutors.remove

移除之前注册的缓存信息

③slotManager.unregisterTaskManager

从slotManager中删除旧的taskManager注册

④workerStarted

当一个worker被启动时回调得到workerType

⑤getTaskExecutorAddress

获取te地址

⑥new WorkerRegistration<>

构建WorkerRegistration

⑦log.info

taskExecutors.put

打印日志在rm上注册tm

放入缓存中

⑧taskManagerHeartbeatManager.monitorTarget

监控tm作为心跳目标

⑨new TaskExecutorRegistrationSuccess

创建并返回注册成功的信息

由于涉及到rpc调用，发送了注册成功的信息，那么就一定会回调TaskExecutor中的onRegistrationSuccess方法，剩下的我们下期分析。

总览

这一期涉及到的源码流程图如下：

我们下期见！