@Test public void enqueuingMinSizeTriggersCheckOnExecutor() { doReturn(1).when(executor).getActiveCount(); doReturn(1).when(executor).getPoolSize(); SingleRollupWriteContext srwc1 = mock(SingleRollupWriteContext.class); SingleRollupWriteContext srwc2 = mock(SingleRollupWriteContext.class); SingleRollupWriteContext srwc3 = mock(SingleRollupWriteContext.class); SingleRollupWriteContext srwc4 = mock(SingleRollupWriteContext.class); SingleRollupWriteContext srwc5 = mock(SingleRollupWriteContext.class); rbw.enqueueRollupForWrite(srwc1); rbw.enqueueRollupForWrite(srwc2); rbw.enqueueRollupForWrite(srwc3); rbw.enqueueRollupForWrite(srwc4); rbw.enqueueRollupForWrite(srwc5); verify(ctx, times(5)).incrementWriteCounter(); verifyNoMoreInteractions(ctx); verifyZeroInteractions(srwc1); verifyZeroInteractions(srwc2); verifyZeroInteractions(srwc3); verifyZeroInteractions(srwc4); verifyZeroInteractions(srwc5); verify(executor).getActiveCount(); verify(executor).getPoolSize(); verifyNoMoreInteractions(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();  }