@Test public void testRequestTaskManagerInfo() throws Exception { final ResourceID taskManagerId = ResourceID.generate(); final TaskExecutorGateway taskExecutorGateway = new TestingTaskExecutorGatewayBuilder().setAddress(UUID.randomUUID().toString()).createTestingTaskExecutorGateway(); rpcService.registerGateway(taskExecutorGateway.getAddress(), taskExecutorGateway); resourceManager = createAndStartResourceManager(heartbeatServices); final ResourceManagerGateway resourceManagerGateway = resourceManager.getSelfGateway(ResourceManagerGateway.class); registerTaskExecutor(resourceManagerGateway, taskManagerId, taskExecutorGateway.getAddress()); CompletableFuture<TaskManagerInfo> taskManagerInfoFuture = resourceManagerGateway.requestTaskManagerInfo( taskManagerId, TestingUtils.TIMEOUT()); TaskManagerInfo taskManagerInfo = taskManagerInfoFuture.get(); Assert.assertEquals(taskManagerId, taskManagerInfo.getResourceId()); Assert.assertEquals(hardwareDescription, taskManagerInfo.getHardwareDescription()); Assert.assertEquals(taskExecutorGateway.getAddress(), taskManagerInfo.getAddress()); Assert.assertEquals(dataPort, taskManagerInfo.getDataPort()); Assert.assertEquals(0, taskManagerInfo.getNumberSlots()); Assert.assertEquals(0, taskManagerInfo.getNumberAvailableSlots()); }
@Override public CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout) { final ArrayList<TaskManagerInfo> taskManagerInfos = new ArrayList<>(taskExecutors.size()); for (Map.Entry<ResourceID, WorkerRegistration<WorkerType>> taskExecutorEntry : taskExecutors.entrySet()) { final ResourceID resourceId = taskExecutorEntry.getKey(); final WorkerRegistration<WorkerType> taskExecutor = taskExecutorEntry.getValue(); taskManagerInfos.add( new TaskManagerInfo( resourceId, taskExecutor.getTaskExecutorGateway().getAddress(), taskExecutor.getDataPort(), taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId), slotManager.getNumberRegisteredSlotsOf(taskExecutor.getInstanceID()), slotManager.getNumberFreeSlotsOf(taskExecutor.getInstanceID()), taskExecutor.getHardwareDescription())); } return CompletableFuture.completedFuture(taskManagerInfos); }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { @Override public CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout) { final ArrayList<TaskManagerInfo> taskManagerInfos = new ArrayList<>(taskExecutors.size()); for (Map.Entry<ResourceID, WorkerRegistration<WorkerType>> taskExecutorEntry : taskExecutors.entrySet()) { final ResourceID resourceId = taskExecutorEntry.getKey(); final WorkerRegistration<WorkerType> taskExecutor = taskExecutorEntry.getValue(); taskManagerInfos.add( new TaskManagerInfo( resourceId, taskExecutor.getTaskExecutorGateway().getAddress(), taskExecutor.getDataPort(), taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId), slotManager.getNumberRegisteredSlotsOf(taskExecutor.getInstanceID()), slotManager.getNumberFreeSlotsOf(taskExecutor.getInstanceID()), taskExecutor.getHardwareDescription())); } return CompletableFuture.completedFuture(taskManagerInfos); } }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { @Override public CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout) { final ArrayList<TaskManagerInfo> taskManagerInfos = new ArrayList<>(taskExecutors.size()); for (Map.Entry<ResourceID, WorkerRegistration<WorkerType>> taskExecutorEntry : taskExecutors.entrySet()) { final ResourceID resourceId = taskExecutorEntry.getKey(); final WorkerRegistration<WorkerType> taskExecutor = taskExecutorEntry.getValue(); taskManagerInfos.add( new TaskManagerInfo( resourceId, taskExecutor.getTaskExecutorGateway().getAddress(), taskExecutor.getDataPort(), taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId), slotManager.getNumberRegisteredSlotsOf(taskExecutor.getInstanceID()), slotManager.getNumberFreeSlotsOf(taskExecutor.getInstanceID()), taskExecutor.getHardwareDescription())); } return CompletableFuture.completedFuture(taskManagerInfos); } 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<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout) { final ArrayList<TaskManagerInfo> taskManagerInfos = new ArrayList<>(taskExecutors.size()); for (Map.Entry<ResourceID, WorkerRegistration<WorkerType>> taskExecutorEntry : taskExecutors.entrySet()) { final ResourceID resourceId = taskExecutorEntry.getKey(); final WorkerRegistration<WorkerType> taskExecutor = taskExecutorEntry.getValue(); taskManagerInfos.add( new TaskManagerInfo( resourceId, taskExecutor.getTaskExecutorGateway().getAddress(), taskExecutor.getDataPort(), taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId), slotManager.getNumberRegisteredSlotsOf(taskExecutor.getInstanceID()), slotManager.getNumberFreeSlotsOf(taskExecutor.getInstanceID()), taskExecutor.getHardwareDescription())); } return CompletableFuture.completedFuture(taskManagerInfos); } 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 void registerInfoMessageListener(final String address); @Override void unRegisterInfoMessageListener(final String address); @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>>> requestTaskManagerMetricQueryServicePaths(Time timeout); @Override CompletableFuture<TransientBlobKey> requestTaskManagerFileUpload(ResourceID taskManagerId, FileType fileType, Time timeout); void sendInfoMessage(final String message); @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); }
ResourceManager extends FencedRpcEndpoint<ResourceManagerId> implements ResourceManagerGateway, LeaderContender { @Override public CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout) { final ArrayList<TaskManagerInfo> taskManagerInfos = new ArrayList<>(taskExecutors.size()); for (Map.Entry<ResourceID, WorkerRegistration<WorkerType>> taskExecutorEntry : taskExecutors.entrySet()) { final ResourceID resourceId = taskExecutorEntry.getKey(); final WorkerRegistration<WorkerType> taskExecutor = taskExecutorEntry.getValue(); taskManagerInfos.add( new TaskManagerInfo( resourceId, taskExecutor.getTaskExecutorGateway().getAddress(), taskExecutor.getDataPort(), taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId), slotManager.getNumberRegisteredSlotsOf(taskExecutor.getInstanceID()), slotManager.getNumberFreeSlotsOf(taskExecutor.getInstanceID()), taskExecutor.getHardwareDescription())); } return CompletableFuture.completedFuture(taskManagerInfos); } 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 void registerInfoMessageListener(final String address); @Override void unRegisterInfoMessageListener(final String address); @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>>> requestTaskManagerMetricQueryServicePaths(Time timeout); @Override CompletableFuture<TransientBlobKey> requestTaskManagerFileUpload(ResourceID taskManagerId, FileType fileType, Time timeout); void sendInfoMessage(final String message); @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 final String RESOURCE_MANAGER_NAME; }