@Test public void executeRollupForLocatorTriggersExecutionOfRollupRunnable() { lfr.executeRollupForLocator(executionContext, rollupBatchWriter, locators.get(0)); verify(rollupReadExecutor, times(1)).execute(Matchers.<RollupRunnable>any()); verifyNoMoreInteractions(rollupReadExecutor); verify(executionContext, times(1)).incrementReadCounter(); verifyNoMoreInteractions(executionContext); verifyZeroInteractions(rollupBatchWriter); }
public void executeRollupForLocator(RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { executionContext.incrementReadCounter(); final SingleRollupReadContext singleRollupReadContext = new SingleRollupReadContext(locator, parentRange, getGranularity()); RollupRunnable rollupRunnable = new RollupRunnable(executionContext, singleRollupReadContext, rollupBatchWriter); rollupReadExecutor.execute(rollupRunnable); }
LocatorFetchRunnable implements Runnable { public void executeRollupForLocator(RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { executionContext.incrementReadCounter(); final SingleRollupReadContext singleRollupReadContext = new SingleRollupReadContext(locator, parentRange, getGranularity()); RollupRunnable rollupRunnable = new RollupRunnable(executionContext, singleRollupReadContext, rollupBatchWriter); rollupReadExecutor.execute(rollupRunnable); } }
LocatorFetchRunnable implements Runnable { public void executeRollupForLocator(RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { executionContext.incrementReadCounter(); final SingleRollupReadContext singleRollupReadContext = new SingleRollupReadContext(locator, parentRange, getGranularity()); RollupRunnable rollupRunnable = new RollupRunnable(executionContext, singleRollupReadContext, rollupBatchWriter); rollupReadExecutor.execute(rollupRunnable); } LocatorFetchRunnable(ScheduleContext scheduleCtx,
                         SlotKey destSlotKey,
                         ExecutorService rollupReadExecutor,
                         ThreadPoolExecutor rollupWriteExecutor); }
LocatorFetchRunnable implements Runnable { public void executeRollupForLocator(RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { executionContext.incrementReadCounter(); final SingleRollupReadContext singleRollupReadContext = new SingleRollupReadContext(locator, parentRange, getGranularity()); RollupRunnable rollupRunnable = new RollupRunnable(executionContext, singleRollupReadContext, rollupBatchWriter); rollupReadExecutor.execute(rollupRunnable); } 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 { public void executeRollupForLocator(RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { executionContext.incrementReadCounter(); final SingleRollupReadContext singleRollupReadContext = new SingleRollupReadContext(locator, parentRange, getGranularity()); RollupRunnable rollupRunnable = new RollupRunnable(executionContext, singleRollupReadContext, rollupBatchWriter); rollupReadExecutor.execute(rollupRunnable); } 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);  }