@Test public void enqueueIncrementsWriterCounter() { SingleRollupWriteContext srwc = mock(SingleRollupWriteContext.class); rbw.enqueueRollupForWrite(srwc); verify(ctx).incrementWriteCounter(); verifyNoMoreInteractions(ctx); verifyZeroInteractions(srwc); }
public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } }
RollupBatchWriter { public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } } }
RollupBatchWriter { public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } } RollupBatchWriter(ThreadPoolExecutor executor, RollupExecutionContext context); }
RollupBatchWriter { public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } } RollupBatchWriter(ThreadPoolExecutor executor, RollupExecutionContext context); void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext); synchronized void drainBatch(); }
RollupBatchWriter { public void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext) { rollupQueue.add(rollupWriteContext); context.incrementWriteCounter(); if (rollupQueue.size() >= ROLLUP_BATCH_MIN_SIZE) { if (executor.getActiveCount() < executor.getPoolSize() || rollupQueue.size() >= ROLLUP_BATCH_MAX_SIZE) { drainBatch(); } } } RollupBatchWriter(ThreadPoolExecutor executor, RollupExecutionContext context); void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext); synchronized void drainBatch();  }