@Test public void enqueuingLessThanMinSizeDoesNotTriggerBatching() { SingleRollupWriteContext srwc1 = mock(SingleRollupWriteContext.class); SingleRollupWriteContext srwc2 = mock(SingleRollupWriteContext.class); SingleRollupWriteContext srwc3 = mock(SingleRollupWriteContext.class); SingleRollupWriteContext srwc4 = mock(SingleRollupWriteContext.class); rbw.enqueueRollupForWrite(srwc1); rbw.enqueueRollupForWrite(srwc2); rbw.enqueueRollupForWrite(srwc3); rbw.enqueueRollupForWrite(srwc4); verify(ctx, times(4)).incrementWriteCounter(); verifyNoMoreInteractions(ctx); verifyZeroInteractions(srwc1); verifyZeroInteractions(srwc2); verifyZeroInteractions(srwc3); verifyZeroInteractions(srwc4); verifyZeroInteractions(executor); }
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();  }