### Question:
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { public void acceptOffers(AcceptOffers msg) { try { List<TaskMonitor.TaskGoalStateUpdated> toMonitor = new ArrayList<>(msg.operations().size()); for (Protos.Offer.Operation op : msg.operations()) { if (op.getType() == Protos.Offer.Operation.Type.LAUNCH) { for (Protos.TaskInfo info : op.getLaunch().getTaskInfosList()) { MesosWorkerStore.Worker worker = workersInNew.remove(extractResourceID(info.getTaskId())); assert (worker != null); worker = worker.launchWorker(info.getSlaveId(), msg.hostname()); workerStore.putWorker(worker); workersInLaunch.put(extractResourceID(worker.taskID()), worker); LOG.info("Launching Mesos task {} on host {}.", worker.taskID().getValue(), worker.hostname().get()); toMonitor.add(new TaskMonitor.TaskGoalStateUpdated(extractGoalState(worker))); } } } for (TaskMonitor.TaskGoalStateUpdated update : toMonitor) { taskMonitor.tell(update, selfActor); } schedulerDriver.acceptOffers(msg.offerIds(), msg.operations(), msg.filters()); } catch (Exception ex) { onFatalError(new ResourceManagerException("unable to accept offers", ex)); } } 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);  }### Answer:
@Test public void testAcceptOffers() throws Exception { new Context() {{ startResourceManager(); MesosWorkerStore.Worker worker1 = allocateWorker(task1, resourceProfile1); Protos.TaskInfo task1info = Protos.TaskInfo.newBuilder() .setTaskId(task1).setName("").setSlaveId(slave1).build(); AcceptOffers msg = new AcceptOffers(slave1host, singletonList(offer1), singletonList(launch(task1info))); resourceManager.acceptOffers(msg); MesosWorkerStore.Worker worker1launched = worker1.launchWorker(slave1, slave1host); verify(rmServices.workerStore).putWorker(worker1launched); assertThat(resourceManager.workersInNew.entrySet(), empty()); assertThat(resourceManager.workersInLaunch, hasEntry(extractResourceID(task1), worker1launched)); resourceManager.taskRouter.expectMsg( new TaskMonitor.TaskGoalStateUpdated(extractGoalState(worker1launched))); verify(rmServices.schedulerDriver).acceptOffers(msg.offerIds(), msg.operations(), msg.filters()); }}; }