MesosSchedulerImpl implements Scheduler { @Timed("scheduler_resource_offers") @Override public void resourceOffers(SchedulerDriver driver, final List<Offer> offers) { Preconditions.checkState(isRegistered, "Must be registered before receiving offers."); executor.execute(() -> { storage.write((NoResult.Quiet) storeProvider -> { for (Offer offer : offers) { IHostAttributes attributes = AttributeStore.Util.mergeOffer(storeProvider.getAttributeStore(), offer); storeProvider.getAttributeStore().saveHostAttributes(attributes); log.debug("Received offer: {}", offer); counters.get("scheduler_resource_offers").incrementAndGet(); offerManager.addOffer(new HostOffer(offer, attributes)); } }); }); } @Inject  MesosSchedulerImpl(
      Storage storage,
      Lifecycle lifecycle,
      TaskStatusHandler taskStatusHandler,
      OfferManager offerManager,
      EventSink eventSink,
      @SchedulerExecutor Executor executor,
      CachedCounters counters); @VisibleForTesting  MesosSchedulerImpl(
      Storage storage,
      Lifecycle lifecycle,
      TaskStatusHandler taskStatusHandler,
      OfferManager offerManager,
      EventSink eventSink,
      Executor executor,
      CachedCounters counters,
      Logger log); @Override void slaveLost(SchedulerDriver schedulerDriver, SlaveID slaveId); @Override void registered(
      SchedulerDriver driver,
      final FrameworkID frameworkId,
      MasterInfo masterInfo); @Override void disconnected(SchedulerDriver schedulerDriver); @Override void reregistered(SchedulerDriver schedulerDriver, MasterInfo masterInfo); @Timed("scheduler_resource_offers") @Override void resourceOffers(SchedulerDriver driver, final List<Offer> offers); @Override void offerRescinded(SchedulerDriver schedulerDriver, OfferID offerId); @AllowUnchecked @Timed("scheduler_status_update") @Override void statusUpdate(SchedulerDriver driver, TaskStatus status); @Override void error(SchedulerDriver driver, String message); @Override void executorLost(SchedulerDriver schedulerDriver, ExecutorID executorID, SlaveID slaveID,
      int status); @Timed("scheduler_framework_message") @Override void frameworkMessage(
      SchedulerDriver driver,
      ExecutorID executorID,
      SlaveID slave,
      byte[] data);  }
@Test(expected = IllegalStateException.class) public void testBadOrdering() { control.replay(); scheduler.resourceOffers(driver, ImmutableList.of()); }
@Test public void testNoOffers() { new AbstractRegisteredTest() { @Override void test() { scheduler.resourceOffers(driver, ImmutableList.of()); } }.run(); }
@Test public void testMultipleOffers() { new AbstractRegisteredTest() { @Override void expectations() { expectOfferAttributesSaved(OFFER); expectOfferAttributesSaved(OFFER_2); offerManager.addOffer(OFFER); offerManager.addOffer(OFFER_2); } @Override void test() { scheduler.resourceOffers(driver, ImmutableList.of(OFFER.getOffer(), OFFER_2.getOffer())); } }.run(); }