@Test public void testWorkerStarted() throws Exception { new Context() {{ MesosWorkerStore.Worker worker1launched = MesosWorkerStore.Worker.newWorker(task1).launchWorker(slave1, slave1host); when(rmServices.workerStore.getFrameworkID()).thenReturn(Option.apply(framework1)); when(rmServices.workerStore.recoverWorkers()).thenReturn(singletonList(worker1launched)); startResourceManager(); assertThat(resourceManager.workersInLaunch, hasEntry(extractResourceID(task1), worker1launched)); final int dataPort = 1234; final HardwareDescription hardwareDescription = new HardwareDescription(1, 2L, 3L, 4L); CompletableFuture<RegistrationResponse> successfulFuture = resourceManager.registerTaskExecutor(task1Executor.address, task1Executor.resourceID, dataPort, hardwareDescription, timeout); RegistrationResponse response = successfulFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS); assertTrue(response instanceof TaskExecutorRegistrationSuccess); final TaskExecutorRegistrationSuccess registrationResponse = (TaskExecutorRegistrationSuccess) response; final CompletableFuture<Acknowledge> initialSlotReportFuture = resourceManager.sendSlotReport(task1Executor.resourceID, registrationResponse.getRegistrationId(), slotReport, timeout); initialSlotReportFuture.get(); assertThat(resourceManager.workersInLaunch, hasEntry(extractResourceID(task1), worker1launched)); }}; }
@Override protected RegisteredMesosWorkerNode workerStarted(ResourceID resourceID) { MesosWorkerStore.Worker inLaunch = workersInLaunch.get(resourceID); if (inLaunch != null) { return new RegisteredMesosWorkerNode(inLaunch); } else { return null; } }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override protected RegisteredMesosWorkerNode workerStarted(ResourceID resourceID) { MesosWorkerStore.Worker inLaunch = workersInLaunch.get(resourceID); if (inLaunch != null) { return new RegisteredMesosWorkerNode(inLaunch); } else { return null; } } }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override protected RegisteredMesosWorkerNode workerStarted(ResourceID resourceID) { MesosWorkerStore.Worker inLaunch = workersInLaunch.get(resourceID); if (inLaunch != null) { return new RegisteredMesosWorkerNode(inLaunch); } else { return null; } } MesosResourceManager(
			// base class
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			// Mesos specifics
			Configuration flinkConfig,
			MesosServices mesosServices,
			MesosConfiguration mesosConfig,
			MesosTaskManagerParameters taskManagerParameters,
			ContainerSpecification taskManagerContainerSpec,
			@Nullable String webUiUrl,
			JobManagerMetricGroup jobManagerMetricGroup); }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override protected RegisteredMesosWorkerNode workerStarted(ResourceID resourceID) { MesosWorkerStore.Worker inLaunch = workersInLaunch.get(resourceID); if (inLaunch != null) { return new RegisteredMesosWorkerNode(inLaunch); } else { return null; } } MesosResourceManager(
			// base class
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			// Mesos specifics
			Configuration flinkConfig,
			MesosServices mesosServices,
			MesosConfiguration mesosConfig,
			MesosTaskManagerParameters taskManagerParameters,
			ContainerSpecification taskManagerContainerSpec,
			@Nullable String webUiUrl,
			JobManagerMetricGroup jobManagerMetricGroup); @Override CompletableFuture<Void> onStop(); @Override Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); @Override boolean stopWorker(RegisteredMesosWorkerNode workerNode); void acceptOffers(AcceptOffers msg); void reconcile(ReconciliationCoordinator.Reconcile message); void taskTerminated(TaskMonitor.TaskTerminated message); }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override protected RegisteredMesosWorkerNode workerStarted(ResourceID resourceID) { MesosWorkerStore.Worker inLaunch = workersInLaunch.get(resourceID); if (inLaunch != null) { return new RegisteredMesosWorkerNode(inLaunch); } else { return null; } } MesosResourceManager(
			// base class
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			// Mesos specifics
			Configuration flinkConfig,
			MesosServices mesosServices,
			MesosConfiguration mesosConfig,
			MesosTaskManagerParameters taskManagerParameters,
			ContainerSpecification taskManagerContainerSpec,
			@Nullable String webUiUrl,
			JobManagerMetricGroup jobManagerMetricGroup); @Override CompletableFuture<Void> onStop(); @Override Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); @Override boolean stopWorker(RegisteredMesosWorkerNode workerNode); void acceptOffers(AcceptOffers msg); void reconcile(ReconciliationCoordinator.Reconcile message); void taskTerminated(TaskMonitor.TaskTerminated message);  }