@Test public void processLocatorExceptionCausesRollupToFail() { Throwable cause = new UnsupportedOperationException("exception for testing purposes"); doThrow(cause).when(rollupReadExecutor).execute(Matchers.<Runnable>any()); int count = lfr.processLocator(0, executionContext, rollupBatchWriter, locators.get(0)); Assert.assertEquals(0, count); verify(executionContext, times(1)).markUnsuccessful(Matchers.<Throwable>any()); verify(executionContext, times(1)).decrementReadCounter(); }
public int processLocator(int rollCount, RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { if (log.isTraceEnabled()) log.trace("Rolling up (check,metric,dimension) {} for (gran,slot,shard) {}", locator, parentSlotKey); try { executeRollupForLocator(executionContext, rollupBatchWriter, locator); rollCount += 1; } catch (Throwable any) { executionContext.markUnsuccessful(any); executionContext.decrementReadCounter(); log.error(String.format( "BasicRollup failed for %s, locator %s, at %d", parentSlotKey, locator, serverTime), any); } return rollCount; }
LocatorFetchRunnable implements Runnable { public int processLocator(int rollCount, RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { if (log.isTraceEnabled()) log.trace("Rolling up (check,metric,dimension) {} for (gran,slot,shard) {}", locator, parentSlotKey); try { executeRollupForLocator(executionContext, rollupBatchWriter, locator); rollCount += 1; } catch (Throwable any) { executionContext.markUnsuccessful(any); executionContext.decrementReadCounter(); log.error(String.format( "BasicRollup failed for %s, locator %s, at %d", parentSlotKey, locator, serverTime), any); } return rollCount; } }
LocatorFetchRunnable implements Runnable { public int processLocator(int rollCount, RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { if (log.isTraceEnabled()) log.trace("Rolling up (check,metric,dimension) {} for (gran,slot,shard) {}", locator, parentSlotKey); try { executeRollupForLocator(executionContext, rollupBatchWriter, locator); rollCount += 1; } catch (Throwable any) { executionContext.markUnsuccessful(any); executionContext.decrementReadCounter(); log.error(String.format( "BasicRollup failed for %s, locator %s, at %d", parentSlotKey, locator, serverTime), any); } return rollCount; } LocatorFetchRunnable(ScheduleContext scheduleCtx,
                         SlotKey destSlotKey,
                         ExecutorService rollupReadExecutor,
                         ThreadPoolExecutor rollupWriteExecutor); }
LocatorFetchRunnable implements Runnable { public int processLocator(int rollCount, RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { if (log.isTraceEnabled()) log.trace("Rolling up (check,metric,dimension) {} for (gran,slot,shard) {}", locator, parentSlotKey); try { executeRollupForLocator(executionContext, rollupBatchWriter, locator); rollCount += 1; } catch (Throwable any) { executionContext.markUnsuccessful(any); executionContext.decrementReadCounter(); log.error(String.format( "BasicRollup failed for %s, locator %s, at %d", parentSlotKey, locator, serverTime), any); } return rollCount; } 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 int processLocator(int rollCount, RollupExecutionContext executionContext, RollupBatchWriter rollupBatchWriter, Locator locator) { if (log.isTraceEnabled()) log.trace("Rolling up (check,metric,dimension) {} for (gran,slot,shard) {}", locator, parentSlotKey); try { executeRollupForLocator(executionContext, rollupBatchWriter, locator); rollCount += 1; } catch (Throwable any) { executionContext.markUnsuccessful(any); executionContext.decrementReadCounter(); log.error(String.format( "BasicRollup failed for %s, locator %s, at %d", parentSlotKey, locator, serverTime), any); } return rollCount; } 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);  }