@Test public void testTierSlaComputationCorrectness() throws Exception { updater.tieredQueueSlaUpdates().subscribe(testSubscriber); testScheduler.triggerActions(); MyTieredQueueSlas slas = (MyTieredQueueSlas) testSubscriber.takeNext(); ResAllocs criticalTierCapacity = slas.getTierCapacities().get(Tier.Critical.ordinal()); assertThat(fromResAllocs(criticalTierCapacity)).isEqualTo(CRITICAL_CAPACITY); ResAllocs criticalGrCapActual = slas.getGroupCapacities().get(Tier.Critical.ordinal()).get(CRITICAL_CAPACITY_GROUP.getAppName()); assertThat(fromResAllocs(criticalGrCapActual)).isEqualTo(toResourceDimension(CRITICAL_CAPACITY_GROUP)); ResAllocs flexTierCapacity = slas.getTierCapacities().get(Tier.Flex.ordinal()); assertThat(fromResAllocs(flexTierCapacity)).isEqualTo(FLEX_CAPACITY); ResAllocs flexGrCapActual = slas.getGroupCapacities().get(Tier.Flex.ordinal()).get(FLEX_CAPACITY_GROUP.getAppName()); assertThat(fromResAllocs(flexGrCapActual)).isEqualTo(toResourceDimension(FLEX_CAPACITY_GROUP)); }
@Override public Observable<TieredQueueSlas> tieredQueueSlaUpdates() { return Observable.interval(0, schedulerConfiguration.getTierSlaUpdateIntervalMs(), TimeUnit.MILLISECONDS, scheduler) .flatMap(tick -> { try { return Observable.just(recomputeSLAs()); } catch (Exception e) { logger.warn("Failed to recompute tier SLAs", e); return Observable.empty(); } }).share(); }
DefaultTierSlaUpdater implements TierSlaUpdater { @Override public Observable<TieredQueueSlas> tieredQueueSlaUpdates() { return Observable.interval(0, schedulerConfiguration.getTierSlaUpdateIntervalMs(), TimeUnit.MILLISECONDS, scheduler) .flatMap(tick -> { try { return Observable.just(recomputeSLAs()); } catch (Exception e) { logger.warn("Failed to recompute tier SLAs", e); return Observable.empty(); } }).share(); } }
DefaultTierSlaUpdater implements TierSlaUpdater { @Override public Observable<TieredQueueSlas> tieredQueueSlaUpdates() { return Observable.interval(0, schedulerConfiguration.getTierSlaUpdateIntervalMs(), TimeUnit.MILLISECONDS, scheduler) .flatMap(tick -> { try { return Observable.just(recomputeSLAs()); } catch (Exception e) { logger.warn("Failed to recompute tier SLAs", e); return Observable.empty(); } }).share(); } @Inject  DefaultTierSlaUpdater(SchedulerConfiguration schedulerConfiguration,
                                 ApplicationSlaManagementService applicationSlaManagementService,
                                 AvailableCapacityService availableCapacityService);   DefaultTierSlaUpdater(SchedulerConfiguration schedulerConfiguration,
                          ApplicationSlaManagementService applicationSlaManagementService,
                          AvailableCapacityService availableCapacityService,
                          Scheduler scheduler); }
DefaultTierSlaUpdater implements TierSlaUpdater { @Override public Observable<TieredQueueSlas> tieredQueueSlaUpdates() { return Observable.interval(0, schedulerConfiguration.getTierSlaUpdateIntervalMs(), TimeUnit.MILLISECONDS, scheduler) .flatMap(tick -> { try { return Observable.just(recomputeSLAs()); } catch (Exception e) { logger.warn("Failed to recompute tier SLAs", e); return Observable.empty(); } }).share(); } @Inject  DefaultTierSlaUpdater(SchedulerConfiguration schedulerConfiguration,
                                 ApplicationSlaManagementService applicationSlaManagementService,
                                 AvailableCapacityService availableCapacityService);   DefaultTierSlaUpdater(SchedulerConfiguration schedulerConfiguration,
                          ApplicationSlaManagementService applicationSlaManagementService,
                          AvailableCapacityService availableCapacityService,
                          Scheduler scheduler); @Override Observable<TieredQueueSlas> tieredQueueSlaUpdates(); }
DefaultTierSlaUpdater implements TierSlaUpdater { @Override public Observable<TieredQueueSlas> tieredQueueSlaUpdates() { return Observable.interval(0, schedulerConfiguration.getTierSlaUpdateIntervalMs(), TimeUnit.MILLISECONDS, scheduler) .flatMap(tick -> { try { return Observable.just(recomputeSLAs()); } catch (Exception e) { logger.warn("Failed to recompute tier SLAs", e); return Observable.empty(); } }).share(); } @Inject  DefaultTierSlaUpdater(SchedulerConfiguration schedulerConfiguration,
                                 ApplicationSlaManagementService applicationSlaManagementService,
                                 AvailableCapacityService availableCapacityService);   DefaultTierSlaUpdater(SchedulerConfiguration schedulerConfiguration,
                          ApplicationSlaManagementService applicationSlaManagementService,
                          AvailableCapacityService availableCapacityService,
                          Scheduler scheduler); @Override Observable<TieredQueueSlas> tieredQueueSlaUpdates();  }