@Test public void getLocatorsExceptionYieldsEmptySet() throws IOException { when(locatorIO.getLocators(TEST_SHARD)).thenThrow(new RuntimeException("")); Set<Locator> actual = lfr.getLocators(executionContext); verify(locatorIO, times(1)).getLocators(TEST_SHARD); verifyNoMoreInteractions(locatorIO); verify(executionContext, times(1)).markUnsuccessful(Matchers.<Throwable>any()); verifyNoMoreInteractions(executionContext); assertNotNull(actual); Assert.assertEquals(0, actual.size()); }
protected Set<Locator> getLocators(RollupExecutionContext executionContext, boolean isReroll, Granularity delayedMetricsRerollGranularity, Granularity delayedMetricsStorageGranularity) { Set<Locator> locators = new HashSet<Locator>(); if (RECORD_DELAYED_METRICS && isReroll && !getGranularity().isCoarser(delayedMetricsRerollGranularity)) { if (getGranularity().isCoarser(delayedMetricsStorageGranularity)) { for (SlotKey slotKey: parentSlotKey.getChildrenKeys(delayedMetricsStorageGranularity)) { locators.addAll(getDelayedLocators(executionContext, slotKey)); } } else { locators = getDelayedLocators(executionContext, parentSlotKey.extrapolate(delayedMetricsStorageGranularity)); } } else { locators = getLocators(executionContext); } return locators; }
LocatorFetchRunnable implements Runnable { protected Set<Locator> getLocators(RollupExecutionContext executionContext, boolean isReroll, Granularity delayedMetricsRerollGranularity, Granularity delayedMetricsStorageGranularity) { Set<Locator> locators = new HashSet<Locator>(); if (RECORD_DELAYED_METRICS && isReroll && !getGranularity().isCoarser(delayedMetricsRerollGranularity)) { if (getGranularity().isCoarser(delayedMetricsStorageGranularity)) { for (SlotKey slotKey: parentSlotKey.getChildrenKeys(delayedMetricsStorageGranularity)) { locators.addAll(getDelayedLocators(executionContext, slotKey)); } } else { locators = getDelayedLocators(executionContext, parentSlotKey.extrapolate(delayedMetricsStorageGranularity)); } } else { locators = getLocators(executionContext); } return locators; } }
LocatorFetchRunnable implements Runnable { protected Set<Locator> getLocators(RollupExecutionContext executionContext, boolean isReroll, Granularity delayedMetricsRerollGranularity, Granularity delayedMetricsStorageGranularity) { Set<Locator> locators = new HashSet<Locator>(); if (RECORD_DELAYED_METRICS && isReroll && !getGranularity().isCoarser(delayedMetricsRerollGranularity)) { if (getGranularity().isCoarser(delayedMetricsStorageGranularity)) { for (SlotKey slotKey: parentSlotKey.getChildrenKeys(delayedMetricsStorageGranularity)) { locators.addAll(getDelayedLocators(executionContext, slotKey)); } } else { locators = getDelayedLocators(executionContext, parentSlotKey.extrapolate(delayedMetricsStorageGranularity)); } } else { locators = getLocators(executionContext); } return locators; } LocatorFetchRunnable(ScheduleContext scheduleCtx,
                         SlotKey destSlotKey,
                         ExecutorService rollupReadExecutor,
                         ThreadPoolExecutor rollupWriteExecutor); }
LocatorFetchRunnable implements Runnable { protected Set<Locator> getLocators(RollupExecutionContext executionContext, boolean isReroll, Granularity delayedMetricsRerollGranularity, Granularity delayedMetricsStorageGranularity) { Set<Locator> locators = new HashSet<Locator>(); if (RECORD_DELAYED_METRICS && isReroll && !getGranularity().isCoarser(delayedMetricsRerollGranularity)) { if (getGranularity().isCoarser(delayedMetricsStorageGranularity)) { for (SlotKey slotKey: parentSlotKey.getChildrenKeys(delayedMetricsStorageGranularity)) { locators.addAll(getDelayedLocators(executionContext, slotKey)); } } else { locators = getDelayedLocators(executionContext, parentSlotKey.extrapolate(delayedMetricsStorageGranularity)); } } else { locators = getLocators(executionContext); } return locators; } LocatorFetchRunnable(ScheduleContext scheduleCtx,
                         SlotKey destSlotKey,
                         ExecutorService rollupReadExecutor,
                         ThreadPoolExecutor rollupWriteExecutor); @VisibleForTesting void initialize(ScheduleContext scheduleCtx,
                           SlotKey destSlotKey,
                           ExecutorService rollupReadExecutor,
                           ThreadPoolExecutor rollupWriteExecutor); void run(); void drainExecutionContext(long waitStart, int rollCount, RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter); void finishExecution(long waitStart, RollupExecutionContext executionContext); int processLocator(int rollCount, RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator); void executeRollupForLocator(RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator); Set<Locator> getDelayedLocators(RollupExecutionContext executionContext, SlotKey slotkey); Set<Locator> getLocators(RollupExecutionContext executionContext); }
LocatorFetchRunnable implements Runnable { protected Set<Locator> getLocators(RollupExecutionContext executionContext, boolean isReroll, Granularity delayedMetricsRerollGranularity, Granularity delayedMetricsStorageGranularity) { Set<Locator> locators = new HashSet<Locator>(); if (RECORD_DELAYED_METRICS && isReroll && !getGranularity().isCoarser(delayedMetricsRerollGranularity)) { if (getGranularity().isCoarser(delayedMetricsStorageGranularity)) { for (SlotKey slotKey: parentSlotKey.getChildrenKeys(delayedMetricsStorageGranularity)) { locators.addAll(getDelayedLocators(executionContext, slotKey)); } } else { locators = getDelayedLocators(executionContext, parentSlotKey.extrapolate(delayedMetricsStorageGranularity)); } } else { locators = getLocators(executionContext); } return locators; } LocatorFetchRunnable(ScheduleContext scheduleCtx,
                         SlotKey destSlotKey,
                         ExecutorService rollupReadExecutor,
                         ThreadPoolExecutor rollupWriteExecutor); @VisibleForTesting void initialize(ScheduleContext scheduleCtx,
                           SlotKey destSlotKey,
                           ExecutorService rollupReadExecutor,
                           ThreadPoolExecutor rollupWriteExecutor); void run(); void drainExecutionContext(long waitStart, int rollCount, RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter); void finishExecution(long waitStart, RollupExecutionContext executionContext); int processLocator(int rollCount, RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator); void executeRollupForLocator(RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator); Set<Locator> getDelayedLocators(RollupExecutionContext executionContext, SlotKey slotkey); Set<Locator> getLocators(RollupExecutionContext executionContext);  }