@Test public void runSendsRollupsToWriterAndDecrementsCount() throws Exception { final AtomicLong decrementCount = new AtomicLong(0); Answer contextAnswer = new Answer() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { decrementCount.set((Long)invocation.getArguments()[0]); return null; } }; doAnswer(contextAnswer).when(ctx).decrementWriteCounter(anyLong()); final Object[] insertRollupsArg = new Object[1]; Answer writerAnswer = new Answer() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { insertRollupsArg[0] = invocation.getArguments()[0]; return null; } }; doAnswer(writerAnswer).when(writer).insertRollups( Matchers.<ArrayList<SingleRollupWriteContext>>any()); rbwr.run(); verify(writer).insertRollups(Matchers.<ArrayList<SingleRollupWriteContext>>any()); assertSame(wcs, insertRollupsArg[0]); verifyNoMoreInteractions(writer); verify(ctx).decrementWriteCounter(anyLong()); assertEquals(wcs.size(), decrementCount.get()); verifyNoMoreInteractions(ctx); }
@Override public void run() { Timer.Context ctx = batchWriteTimer.time(); try { metricsRW.insertRollups(writeContexts); } catch (Exception e) { LOG.warn("not able to insert rollups", e); executionContext.markUnsuccessful(e); } finally { executionContext.decrementWriteCounter(writeContexts.size()); rollupsPerBatch.update(writeContexts.size()); rollupsWriteRate.mark(writeContexts.size()); RollupService.lastRollupTime.set(System.currentTimeMillis()); ctx.stop(); } }
RollupBatchWriteRunnable implements Runnable { @Override public void run() { Timer.Context ctx = batchWriteTimer.time(); try { metricsRW.insertRollups(writeContexts); } catch (Exception e) { LOG.warn("not able to insert rollups", e); executionContext.markUnsuccessful(e); } finally { executionContext.decrementWriteCounter(writeContexts.size()); rollupsPerBatch.update(writeContexts.size()); rollupsWriteRate.mark(writeContexts.size()); RollupService.lastRollupTime.set(System.currentTimeMillis()); ctx.stop(); } } }
RollupBatchWriteRunnable implements Runnable { @Override public void run() { Timer.Context ctx = batchWriteTimer.time(); try { metricsRW.insertRollups(writeContexts); } catch (Exception e) { LOG.warn("not able to insert rollups", e); executionContext.markUnsuccessful(e); } finally { executionContext.decrementWriteCounter(writeContexts.size()); rollupsPerBatch.update(writeContexts.size()); rollupsWriteRate.mark(writeContexts.size()); RollupService.lastRollupTime.set(System.currentTimeMillis()); ctx.stop(); } } RollupBatchWriteRunnable(List<SingleRollupWriteContext> writeContexts,
                                    RollupExecutionContext executionContext,
                                    AbstractMetricsRW metricsRW); }
RollupBatchWriteRunnable implements Runnable { @Override public void run() { Timer.Context ctx = batchWriteTimer.time(); try { metricsRW.insertRollups(writeContexts); } catch (Exception e) { LOG.warn("not able to insert rollups", e); executionContext.markUnsuccessful(e); } finally { executionContext.decrementWriteCounter(writeContexts.size()); rollupsPerBatch.update(writeContexts.size()); rollupsWriteRate.mark(writeContexts.size()); RollupService.lastRollupTime.set(System.currentTimeMillis()); ctx.stop(); } } RollupBatchWriteRunnable(List<SingleRollupWriteContext> writeContexts,
                                    RollupExecutionContext executionContext,
                                    AbstractMetricsRW metricsRW); @Override void run(); }
RollupBatchWriteRunnable implements Runnable { @Override public void run() { Timer.Context ctx = batchWriteTimer.time(); try { metricsRW.insertRollups(writeContexts); } catch (Exception e) { LOG.warn("not able to insert rollups", e); executionContext.markUnsuccessful(e); } finally { executionContext.decrementWriteCounter(writeContexts.size()); rollupsPerBatch.update(writeContexts.size()); rollupsWriteRate.mark(writeContexts.size()); RollupService.lastRollupTime.set(System.currentTimeMillis()); ctx.stop(); } } RollupBatchWriteRunnable(List<SingleRollupWriteContext> writeContexts,
                                    RollupExecutionContext executionContext,
                                    AbstractMetricsRW metricsRW); @Override void run();  }