@Test public void testStatusHandling() throws Exception { new Context() {{ startResourceManager(); resourceManager.statusUpdate(new StatusUpdate(Protos.TaskStatus.newBuilder() .setTaskId(task1).setSlaveId(slave1).setState(Protos.TaskState.TASK_LOST).build())); resourceManager.reconciliationCoordinator.expectMsgClass(StatusUpdate.class); resourceManager.taskRouter.expectMsgClass(StatusUpdate.class); }}; }
protected void statusUpdate(StatusUpdate message) { taskMonitor.tell(message, selfActor); reconciliationCoordinator.tell(message, selfActor); schedulerDriver.acknowledgeStatusUpdate(message.status()); }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { protected void statusUpdate(StatusUpdate message) { taskMonitor.tell(message, selfActor); reconciliationCoordinator.tell(message, selfActor); schedulerDriver.acknowledgeStatusUpdate(message.status()); } }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { protected void statusUpdate(StatusUpdate message) { taskMonitor.tell(message, selfActor); reconciliationCoordinator.tell(message, selfActor); schedulerDriver.acknowledgeStatusUpdate(message.status()); } 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> { protected void statusUpdate(StatusUpdate message) { taskMonitor.tell(message, selfActor); reconciliationCoordinator.tell(message, selfActor); schedulerDriver.acknowledgeStatusUpdate(message.status()); } 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> { protected void statusUpdate(StatusUpdate message) { taskMonitor.tell(message, selfActor); reconciliationCoordinator.tell(message, selfActor); schedulerDriver.acknowledgeStatusUpdate(message.status()); } 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);  }