MesosSchedulerImpl implements Scheduler { @AllowUnchecked @Timed("scheduler_status_update") @Override public void statusUpdate(SchedulerDriver driver, TaskStatus status) { logStatusUpdate(log, status); eventSink.post(new TaskStatusReceived( status.getState(), Optional.fromNullable(status.getSource()), status.hasReason() ? Optional.of(status.getReason()) : Optional.absent(), Optional.fromNullable(status.getTimestamp()).transform(SECONDS_TO_MICROS))); try { taskStatusHandler.statusUpdate(status); } catch (SchedulerException e) { log.error("Status update failed due to scheduler exception: " + e, e); throw e; } } @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 public void testStatusUpdate() { new StatusUpdater(STATUS).run(); control.verify(); control.reset(); new StatusUpdater(STATUS.toBuilder().clearSource().build()).run(); control.verify(); control.reset(); new StatusUpdater(STATUS.toBuilder().clearReason().build()).run(); control.verify(); control.reset(); new StatusUpdater(STATUS.toBuilder().clearMessage().build()).run(); }
@Test(expected = SchedulerException.class) public void testStatusUpdateFails() { new AbstractStatusTest() { @Override void expectations() { eventSink.post(new TaskStatusReceived( STATUS.getState(), Optional.of(STATUS.getSource()), Optional.of(STATUS.getReason()), Optional.of(1000000L) )); statusHandler.statusUpdate(status); expectLastCall().andThrow(new StorageException("Injected.")); } }.run(); }
@Test public void testStatusReconciliationAcceptsDebugLogging() { Logger mockLogger = createMock(Logger.class); mockLogger.info(anyString()); mockLogger.debug(anyString()); initializeScheduler(mockLogger); new AbstractStatusReconciliationTest() { @Override void expectations() { eventSink.post(PUBSUB_RECONCILIATION_EVENT); statusHandler.statusUpdate(status); } }.run(); }