@Test public void testHeartbeatTimeoutWithJobMaster() throws Exception { final CompletableFuture<ResourceID> heartbeatRequestFuture = new CompletableFuture<>(); final CompletableFuture<ResourceManagerId> disconnectFuture = new CompletableFuture<>(); final TestingJobMasterGateway jobMasterGateway = new TestingJobMasterGatewayBuilder() .setResourceManagerHeartbeatConsumer(heartbeatRequestFuture::complete) .setDisconnectResourceManagerConsumer(disconnectFuture::complete) .build(); rpcService.registerGateway(jobMasterGateway.getAddress(), jobMasterGateway); final JobID jobId = new JobID(); final ResourceID jobMasterResourceId = ResourceID.generate(); final LeaderRetrievalService jobMasterLeaderRetrievalService = new SettableLeaderRetrievalService(jobMasterGateway.getAddress(), jobMasterGateway.getFencingToken().toUUID()); highAvailabilityServices.setJobMasterLeaderRetrieverFunction(requestedJobId -> { assertThat(requestedJobId, is(equalTo(jobId))); return jobMasterLeaderRetrievalService; }); runHeartbeatTimeoutTest( resourceManagerGateway -> { final CompletableFuture<RegistrationResponse> registrationFuture = resourceManagerGateway.registerJobManager( jobMasterGateway.getFencingToken(), jobMasterResourceId, jobMasterGateway.getAddress(), jobId, TIMEOUT); assertThat(registrationFuture.get(), instanceOf(RegistrationResponse.Success.class)); }, resourceManagerResourceId -> { final ResourceID optionalHeartbeatRequestOrigin = heartbeatRequestFuture.getNow(null); assertThat(optionalHeartbeatRequestOrigin, anyOf(is(resourceManagerResourceId), is(nullValue()))); assertThat(disconnectFuture.get(), is(equalTo(resourceManagerId))); }); }
@Override public CompletableFuture<RegistrationResponse> registerJobManager( final JobMasterId jobMasterId, final ResourceID jobManagerResourceId, final String jobManagerAddress, final JobID jobId, final Time timeout) { checkNotNull(jobMasterId); checkNotNull(jobManagerResourceId); checkNotNull(jobManagerAddress); checkNotNull(jobId); if (!jobLeaderIdService.containsJob(jobId)) { try { jobLeaderIdService.addJob(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Could not add the job " + jobId + " to the job id leader service.", e); onFatalError(exception); log.error("Could not add job {} to job leader id service.", jobId, e); return FutureUtils.completedExceptionally(exception); } } log.info("Registering job manager {}@{} for job {}.", jobMasterId, jobManagerAddress, jobId); CompletableFuture<JobMasterId> jobMasterIdFuture; try { jobMasterIdFuture = jobLeaderIdService.getLeaderId(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Cannot obtain the " + "job leader id future to verify the correct job leader.", e); onFatalError(exception); log.debug("Could not obtain the job leader id future to verify the correct job leader."); return FutureUtils.completedExceptionally(exception); } CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getRpcService().connect(jobManagerAddress, jobMasterId, JobMasterGateway.class); CompletableFuture<RegistrationResponse> registrationResponseFuture = jobMasterGatewayFuture.thenCombineAsync( jobMasterIdFuture, (JobMasterGateway jobMasterGateway, JobMasterId leadingJobMasterId) -> { if (Objects.equals(leadingJobMasterId, jobMasterId)) { return registerJobMasterInternal( jobMasterGateway, jobId, jobManagerAddress, jobManagerResourceId); } else { final String declineMessage = String.format( "The leading JobMaster id %s did not match the received JobMaster id %s. " + "This indicates that a JobMaster leader change has happened.", leadingJobMasterId, jobMasterId); log.debug(declineMessage); return new RegistrationResponse.Decline(declineMessage); } }, getMainThreadExecutor()); return registrationResponseFuture.handleAsync( (RegistrationResponse registrationResponse, Throwable throwable) -> { if (throwable != null) { if (log.isDebugEnabled()) { log.debug("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress, throwable); } else { log.info("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress); } return new RegistrationResponse.Decline(throwable.getMessage()); } else { return registrationResponse; } }, getRpcService().getExecutor()); }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { @Override public CompletableFuture<RegistrationResponse> registerJobManager( final JobMasterId jobMasterId, final ResourceID jobManagerResourceId, final String jobManagerAddress, final JobID jobId, final Time timeout) { checkNotNull(jobMasterId); checkNotNull(jobManagerResourceId); checkNotNull(jobManagerAddress); checkNotNull(jobId); if (!jobLeaderIdService.containsJob(jobId)) { try { jobLeaderIdService.addJob(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Could not add the job " + jobId + " to the job id leader service.", e); onFatalError(exception); log.error("Could not add job {} to job leader id service.", jobId, e); return FutureUtils.completedExceptionally(exception); } } log.info("Registering job manager {}@{} for job {}.", jobMasterId, jobManagerAddress, jobId); CompletableFuture<JobMasterId> jobMasterIdFuture; try { jobMasterIdFuture = jobLeaderIdService.getLeaderId(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Cannot obtain the " + "job leader id future to verify the correct job leader.", e); onFatalError(exception); log.debug("Could not obtain the job leader id future to verify the correct job leader."); return FutureUtils.completedExceptionally(exception); } CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getRpcService().connect(jobManagerAddress, jobMasterId, JobMasterGateway.class); CompletableFuture<RegistrationResponse> registrationResponseFuture = jobMasterGatewayFuture.thenCombineAsync( jobMasterIdFuture, (JobMasterGateway jobMasterGateway, JobMasterId leadingJobMasterId) -> { if (Objects.equals(leadingJobMasterId, jobMasterId)) { return registerJobMasterInternal( jobMasterGateway, jobId, jobManagerAddress, jobManagerResourceId); } else { final String declineMessage = String.format( "The leading JobMaster id %s did not match the received JobMaster id %s. " + "This indicates that a JobMaster leader change has happened.", leadingJobMasterId, jobMasterId); log.debug(declineMessage); return new RegistrationResponse.Decline(declineMessage); } }, getMainThreadExecutor()); return registrationResponseFuture.handleAsync( (RegistrationResponse registrationResponse, Throwable throwable) -> { if (throwable != null) { if (log.isDebugEnabled()) { log.debug("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress, throwable); } else { log.info("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress); } return new RegistrationResponse.Decline(throwable.getMessage()); } else { return registrationResponse; } }, getRpcService().getExecutor()); } }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { @Override public CompletableFuture<RegistrationResponse> registerJobManager( final JobMasterId jobMasterId, final ResourceID jobManagerResourceId, final String jobManagerAddress, final JobID jobId, final Time timeout) { checkNotNull(jobMasterId); checkNotNull(jobManagerResourceId); checkNotNull(jobManagerAddress); checkNotNull(jobId); if (!jobLeaderIdService.containsJob(jobId)) { try { jobLeaderIdService.addJob(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Could not add the job " + jobId + " to the job id leader service.", e); onFatalError(exception); log.error("Could not add job {} to job leader id service.", jobId, e); return FutureUtils.completedExceptionally(exception); } } log.info("Registering job manager {}@{} for job {}.", jobMasterId, jobManagerAddress, jobId); CompletableFuture<JobMasterId> jobMasterIdFuture; try { jobMasterIdFuture = jobLeaderIdService.getLeaderId(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Cannot obtain the " + "job leader id future to verify the correct job leader.", e); onFatalError(exception); log.debug("Could not obtain the job leader id future to verify the correct job leader."); return FutureUtils.completedExceptionally(exception); } CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getRpcService().connect(jobManagerAddress, jobMasterId, JobMasterGateway.class); CompletableFuture<RegistrationResponse> registrationResponseFuture = jobMasterGatewayFuture.thenCombineAsync( jobMasterIdFuture, (JobMasterGateway jobMasterGateway, JobMasterId leadingJobMasterId) -> { if (Objects.equals(leadingJobMasterId, jobMasterId)) { return registerJobMasterInternal( jobMasterGateway, jobId, jobManagerAddress, jobManagerResourceId); } else { final String declineMessage = String.format( "The leading JobMaster id %s did not match the received JobMaster id %s. " + "This indicates that a JobMaster leader change has happened.", leadingJobMasterId, jobMasterId); log.debug(declineMessage); return new RegistrationResponse.Decline(declineMessage); } }, getMainThreadExecutor()); return registrationResponseFuture.handleAsync( (RegistrationResponse registrationResponse, Throwable throwable) -> { if (throwable != null) { if (log.isDebugEnabled()) { log.debug("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress, throwable); } else { log.info("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress); } return new RegistrationResponse.Decline(throwable.getMessage()); } else { return registrationResponse; } }, getRpcService().getExecutor()); } ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup); }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { @Override public CompletableFuture<RegistrationResponse> registerJobManager( final JobMasterId jobMasterId, final ResourceID jobManagerResourceId, final String jobManagerAddress, final JobID jobId, final Time timeout) { checkNotNull(jobMasterId); checkNotNull(jobManagerResourceId); checkNotNull(jobManagerAddress); checkNotNull(jobId); if (!jobLeaderIdService.containsJob(jobId)) { try { jobLeaderIdService.addJob(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Could not add the job " + jobId + " to the job id leader service.", e); onFatalError(exception); log.error("Could not add job {} to job leader id service.", jobId, e); return FutureUtils.completedExceptionally(exception); } } log.info("Registering job manager {}@{} for job {}.", jobMasterId, jobManagerAddress, jobId); CompletableFuture<JobMasterId> jobMasterIdFuture; try { jobMasterIdFuture = jobLeaderIdService.getLeaderId(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Cannot obtain the " + "job leader id future to verify the correct job leader.", e); onFatalError(exception); log.debug("Could not obtain the job leader id future to verify the correct job leader."); return FutureUtils.completedExceptionally(exception); } CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getRpcService().connect(jobManagerAddress, jobMasterId, JobMasterGateway.class); CompletableFuture<RegistrationResponse> registrationResponseFuture = jobMasterGatewayFuture.thenCombineAsync( jobMasterIdFuture, (JobMasterGateway jobMasterGateway, JobMasterId leadingJobMasterId) -> { if (Objects.equals(leadingJobMasterId, jobMasterId)) { return registerJobMasterInternal( jobMasterGateway, jobId, jobManagerAddress, jobManagerResourceId); } else { final String declineMessage = String.format( "The leading JobMaster id %s did not match the received JobMaster id %s. " + "This indicates that a JobMaster leader change has happened.", leadingJobMasterId, jobMasterId); log.debug(declineMessage); return new RegistrationResponse.Decline(declineMessage); } }, getMainThreadExecutor()); return registrationResponseFuture.handleAsync( (RegistrationResponse registrationResponse, Throwable throwable) -> { if (throwable != null) { if (log.isDebugEnabled()) { log.debug("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress, throwable); } else { log.info("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress); } return new RegistrationResponse.Decline(throwable.getMessage()); } else { return registrationResponse; } }, getRpcService().getExecutor()); } ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<RegistrationResponse> registerJobManager(
			final JobMasterId jobMasterId,
			final ResourceID jobManagerResourceId,
			final String jobManagerAddress,
			final JobID jobId,
			final Time timeout); @Override CompletableFuture<RegistrationResponse> registerTaskExecutor(
			final String taskExecutorAddress,
			final ResourceID taskExecutorResourceId,
			final int dataPort,
			final HardwareDescription hardwareDescription,
			final Time timeout); @Override CompletableFuture<Acknowledge> sendSlotReport(ResourceID taskManagerResourceId, InstanceID taskManagerRegistrationId, SlotReport slotReport, Time timeout); @Override void heartbeatFromTaskManager(final ResourceID resourceID, final SlotReport slotReport); @Override void heartbeatFromJobManager(final ResourceID resourceID); @Override void disconnectTaskManager(final ResourceID resourceId, final Exception cause); @Override void disconnectJobManager(final JobID jobId, final Exception cause); @Override CompletableFuture<Acknowledge> requestSlot(
			JobMasterId jobMasterId,
			SlotRequest slotRequest,
			final Time timeout); @Override void cancelSlotRequest(AllocationID allocationID); @Override void notifySlotAvailable(
			final InstanceID instanceID,
			final SlotID slotId,
			final AllocationID allocationId); @Override CompletableFuture<Acknowledge> deregisterApplication(
			final ApplicationStatus finalStatus,
			@Nullable final String diagnostics); @Override CompletableFuture<Integer> getNumberOfRegisteredTaskManagers(); @Override CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout); @Override CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(ResourceID resourceId, Time timeout); @Override CompletableFuture<ResourceOverview> requestResourceOverview(Time timeout); @Override CompletableFuture<Collection<Tuple2<ResourceID, String>>> requestTaskManagerMetricQueryServiceAddresses(Time timeout); @Override CompletableFuture<TransientBlobKey> requestTaskManagerFileUpload(ResourceID taskManagerId, FileType fileType, Time timeout); @Override void grantLeadership(final UUID newLeaderSessionID); @Override void revokeLeadership(); @Override void handleError(final Exception exception); @VisibleForTesting abstract Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); abstract boolean stopWorker(WorkerType worker); static Collection<ResourceProfile> createWorkerSlotProfiles(Configuration config); }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { @Override public CompletableFuture<RegistrationResponse> registerJobManager( final JobMasterId jobMasterId, final ResourceID jobManagerResourceId, final String jobManagerAddress, final JobID jobId, final Time timeout) { checkNotNull(jobMasterId); checkNotNull(jobManagerResourceId); checkNotNull(jobManagerAddress); checkNotNull(jobId); if (!jobLeaderIdService.containsJob(jobId)) { try { jobLeaderIdService.addJob(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Could not add the job " + jobId + " to the job id leader service.", e); onFatalError(exception); log.error("Could not add job {} to job leader id service.", jobId, e); return FutureUtils.completedExceptionally(exception); } } log.info("Registering job manager {}@{} for job {}.", jobMasterId, jobManagerAddress, jobId); CompletableFuture<JobMasterId> jobMasterIdFuture; try { jobMasterIdFuture = jobLeaderIdService.getLeaderId(jobId); } catch (Exception e) { ResourceManagerException exception = new ResourceManagerException("Cannot obtain the " + "job leader id future to verify the correct job leader.", e); onFatalError(exception); log.debug("Could not obtain the job leader id future to verify the correct job leader."); return FutureUtils.completedExceptionally(exception); } CompletableFuture<JobMasterGateway> jobMasterGatewayFuture = getRpcService().connect(jobManagerAddress, jobMasterId, JobMasterGateway.class); CompletableFuture<RegistrationResponse> registrationResponseFuture = jobMasterGatewayFuture.thenCombineAsync( jobMasterIdFuture, (JobMasterGateway jobMasterGateway, JobMasterId leadingJobMasterId) -> { if (Objects.equals(leadingJobMasterId, jobMasterId)) { return registerJobMasterInternal( jobMasterGateway, jobId, jobManagerAddress, jobManagerResourceId); } else { final String declineMessage = String.format( "The leading JobMaster id %s did not match the received JobMaster id %s. " + "This indicates that a JobMaster leader change has happened.", leadingJobMasterId, jobMasterId); log.debug(declineMessage); return new RegistrationResponse.Decline(declineMessage); } }, getMainThreadExecutor()); return registrationResponseFuture.handleAsync( (RegistrationResponse registrationResponse, Throwable throwable) -> { if (throwable != null) { if (log.isDebugEnabled()) { log.debug("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress, throwable); } else { log.info("Registration of job manager {}@{} failed.", jobMasterId, jobManagerAddress); } return new RegistrationResponse.Decline(throwable.getMessage()); } else { return registrationResponse; } }, getRpcService().getExecutor()); } ResourceManager(
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			JobManagerMetricGroup jobManagerMetricGroup); @Override void onStart(); @Override CompletableFuture<Void> onStop(); @Override CompletableFuture<RegistrationResponse> registerJobManager(
			final JobMasterId jobMasterId,
			final ResourceID jobManagerResourceId,
			final String jobManagerAddress,
			final JobID jobId,
			final Time timeout); @Override CompletableFuture<RegistrationResponse> registerTaskExecutor(
			final String taskExecutorAddress,
			final ResourceID taskExecutorResourceId,
			final int dataPort,
			final HardwareDescription hardwareDescription,
			final Time timeout); @Override CompletableFuture<Acknowledge> sendSlotReport(ResourceID taskManagerResourceId, InstanceID taskManagerRegistrationId, SlotReport slotReport, Time timeout); @Override void heartbeatFromTaskManager(final ResourceID resourceID, final SlotReport slotReport); @Override void heartbeatFromJobManager(final ResourceID resourceID); @Override void disconnectTaskManager(final ResourceID resourceId, final Exception cause); @Override void disconnectJobManager(final JobID jobId, final Exception cause); @Override CompletableFuture<Acknowledge> requestSlot(
			JobMasterId jobMasterId,
			SlotRequest slotRequest,
			final Time timeout); @Override void cancelSlotRequest(AllocationID allocationID); @Override void notifySlotAvailable(
			final InstanceID instanceID,
			final SlotID slotId,
			final AllocationID allocationId); @Override CompletableFuture<Acknowledge> deregisterApplication(
			final ApplicationStatus finalStatus,
			@Nullable final String diagnostics); @Override CompletableFuture<Integer> getNumberOfRegisteredTaskManagers(); @Override CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout); @Override CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(ResourceID resourceId, Time timeout); @Override CompletableFuture<ResourceOverview> requestResourceOverview(Time timeout); @Override CompletableFuture<Collection<Tuple2<ResourceID, String>>> requestTaskManagerMetricQueryServiceAddresses(Time timeout); @Override CompletableFuture<TransientBlobKey> requestTaskManagerFileUpload(ResourceID taskManagerId, FileType fileType, Time timeout); @Override void grantLeadership(final UUID newLeaderSessionID); @Override void revokeLeadership(); @Override void handleError(final Exception exception); @VisibleForTesting abstract Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); abstract boolean stopWorker(WorkerType worker); static Collection<ResourceProfile> createWorkerSlotProfiles(Configuration config); static final String RESOURCE_MANAGER_NAME; }