@Test public void finishExecutionWhenSuccessful() { when(executionContext.wasSuccessful()).thenReturn(true); lfr.finishExecution(0, executionContext); verify(executionContext, times(1)).wasSuccessful(); verifyNoMoreInteractions(executionContext); verify(scheduleCtx, times(1)).clearFromRunning(Matchers.<SlotKey>any()); verify(scheduleCtx).getCurrentTimeMillis(); verifyNoMoreInteractions(scheduleCtx); }
public void finishExecution(long waitStart, RollupExecutionContext executionContext) { if (executionContext.wasSuccessful()) { this.scheduleCtx.clearFromRunning(parentSlotKey); log.info("Successful completion of rollups for (gran,slot,shard) {} in {} ms", new Object[]{parentSlotKey, System.currentTimeMillis() - waitStart}); } else { log.error("Performing BasicRollups for {} failed", parentSlotKey); this.scheduleCtx.pushBackToScheduled(parentSlotKey, false); } }
LocatorFetchRunnable implements Runnable { public void finishExecution(long waitStart, RollupExecutionContext executionContext) { if (executionContext.wasSuccessful()) { this.scheduleCtx.clearFromRunning(parentSlotKey); log.info("Successful completion of rollups for (gran,slot,shard) {} in {} ms", new Object[]{parentSlotKey, System.currentTimeMillis() - waitStart}); } else { log.error("Performing BasicRollups for {} failed", parentSlotKey); this.scheduleCtx.pushBackToScheduled(parentSlotKey, false); } } }
LocatorFetchRunnable implements Runnable { public void finishExecution(long waitStart, RollupExecutionContext executionContext) { if (executionContext.wasSuccessful()) { this.scheduleCtx.clearFromRunning(parentSlotKey); log.info("Successful completion of rollups for (gran,slot,shard) {} in {} ms", new Object[]{parentSlotKey, System.currentTimeMillis() - waitStart}); } else { log.error("Performing BasicRollups for {} failed", parentSlotKey); this.scheduleCtx.pushBackToScheduled(parentSlotKey, false); } } LocatorFetchRunnable(ScheduleContext scheduleCtx,
                         SlotKey destSlotKey,
                         ExecutorService rollupReadExecutor,
                         ThreadPoolExecutor rollupWriteExecutor); }
LocatorFetchRunnable implements Runnable { public void finishExecution(long waitStart, RollupExecutionContext executionContext) { if (executionContext.wasSuccessful()) { this.scheduleCtx.clearFromRunning(parentSlotKey); log.info("Successful completion of rollups for (gran,slot,shard) {} in {} ms", new Object[]{parentSlotKey, System.currentTimeMillis() - waitStart}); } else { log.error("Performing BasicRollups for {} failed", parentSlotKey); this.scheduleCtx.pushBackToScheduled(parentSlotKey, false); } } 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 finishExecution(long waitStart, RollupExecutionContext executionContext) { if (executionContext.wasSuccessful()) { this.scheduleCtx.clearFromRunning(parentSlotKey); log.info("Successful completion of rollups for (gran,slot,shard) {} in {} ms", new Object[]{parentSlotKey, System.currentTimeMillis() - waitStart}); } else { log.error("Performing BasicRollups for {} failed", parentSlotKey); this.scheduleCtx.pushBackToScheduled(parentSlotKey, false); } } 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);  }