@Test public void drainBatchWithNoItemsDoesNotTriggerBatching() { rbw.drainBatch(); verifyZeroInteractions(ctx); verifyZeroInteractions(executor); }
public synchronized void drainBatch() { List<SingleRollupWriteContext> writeBasicContexts = new ArrayList<SingleRollupWriteContext>(); List<SingleRollupWriteContext> writePreAggrContexts = new ArrayList<SingleRollupWriteContext>(); try { for (int i=0; i<=ROLLUP_BATCH_MAX_SIZE; i++) { SingleRollupWriteContext context = rollupQueue.remove(); if ( context.getRollup().getRollupType() == RollupType.BF_BASIC ) { writeBasicContexts.add(context); } else { writePreAggrContexts.add(context); } } } catch (NoSuchElementException e) { } if (writeBasicContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d basic contexts", writeBasicContexts.size())); executor.execute(new RollupBatchWriteRunnable(writeBasicContexts, context, basicMetricsRW)); } if (writePreAggrContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d preAggr contexts", writePreAggrContexts.size())); executor.execute(new RollupBatchWriteRunnable(writePreAggrContexts, context, preAggregatedRW)); } }
RollupBatchWriter { public synchronized void drainBatch() { List<SingleRollupWriteContext> writeBasicContexts = new ArrayList<SingleRollupWriteContext>(); List<SingleRollupWriteContext> writePreAggrContexts = new ArrayList<SingleRollupWriteContext>(); try { for (int i=0; i<=ROLLUP_BATCH_MAX_SIZE; i++) { SingleRollupWriteContext context = rollupQueue.remove(); if ( context.getRollup().getRollupType() == RollupType.BF_BASIC ) { writeBasicContexts.add(context); } else { writePreAggrContexts.add(context); } } } catch (NoSuchElementException e) { } if (writeBasicContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d basic contexts", writeBasicContexts.size())); executor.execute(new RollupBatchWriteRunnable(writeBasicContexts, context, basicMetricsRW)); } if (writePreAggrContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d preAggr contexts", writePreAggrContexts.size())); executor.execute(new RollupBatchWriteRunnable(writePreAggrContexts, context, preAggregatedRW)); } } }
RollupBatchWriter { public synchronized void drainBatch() { List<SingleRollupWriteContext> writeBasicContexts = new ArrayList<SingleRollupWriteContext>(); List<SingleRollupWriteContext> writePreAggrContexts = new ArrayList<SingleRollupWriteContext>(); try { for (int i=0; i<=ROLLUP_BATCH_MAX_SIZE; i++) { SingleRollupWriteContext context = rollupQueue.remove(); if ( context.getRollup().getRollupType() == RollupType.BF_BASIC ) { writeBasicContexts.add(context); } else { writePreAggrContexts.add(context); } } } catch (NoSuchElementException e) { } if (writeBasicContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d basic contexts", writeBasicContexts.size())); executor.execute(new RollupBatchWriteRunnable(writeBasicContexts, context, basicMetricsRW)); } if (writePreAggrContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d preAggr contexts", writePreAggrContexts.size())); executor.execute(new RollupBatchWriteRunnable(writePreAggrContexts, context, preAggregatedRW)); } } RollupBatchWriter(ThreadPoolExecutor executor, RollupExecutionContext context); }
RollupBatchWriter { public synchronized void drainBatch() { List<SingleRollupWriteContext> writeBasicContexts = new ArrayList<SingleRollupWriteContext>(); List<SingleRollupWriteContext> writePreAggrContexts = new ArrayList<SingleRollupWriteContext>(); try { for (int i=0; i<=ROLLUP_BATCH_MAX_SIZE; i++) { SingleRollupWriteContext context = rollupQueue.remove(); if ( context.getRollup().getRollupType() == RollupType.BF_BASIC ) { writeBasicContexts.add(context); } else { writePreAggrContexts.add(context); } } } catch (NoSuchElementException e) { } if (writeBasicContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d basic contexts", writeBasicContexts.size())); executor.execute(new RollupBatchWriteRunnable(writeBasicContexts, context, basicMetricsRW)); } if (writePreAggrContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d preAggr contexts", writePreAggrContexts.size())); executor.execute(new RollupBatchWriteRunnable(writePreAggrContexts, context, preAggregatedRW)); } } RollupBatchWriter(ThreadPoolExecutor executor, RollupExecutionContext context); void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext); synchronized void drainBatch(); }
RollupBatchWriter { public synchronized void drainBatch() { List<SingleRollupWriteContext> writeBasicContexts = new ArrayList<SingleRollupWriteContext>(); List<SingleRollupWriteContext> writePreAggrContexts = new ArrayList<SingleRollupWriteContext>(); try { for (int i=0; i<=ROLLUP_BATCH_MAX_SIZE; i++) { SingleRollupWriteContext context = rollupQueue.remove(); if ( context.getRollup().getRollupType() == RollupType.BF_BASIC ) { writeBasicContexts.add(context); } else { writePreAggrContexts.add(context); } } } catch (NoSuchElementException e) { } if (writeBasicContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d basic contexts", writeBasicContexts.size())); executor.execute(new RollupBatchWriteRunnable(writeBasicContexts, context, basicMetricsRW)); } if (writePreAggrContexts.size() > 0) { LOG.debug( String.format("drainBatch(): kicking off RollupBatchWriteRunnables for %d preAggr contexts", writePreAggrContexts.size())); executor.execute(new RollupBatchWriteRunnable(writePreAggrContexts, context, preAggregatedRW)); } } RollupBatchWriter(ThreadPoolExecutor executor, RollupExecutionContext context); void enqueueRollupForWrite(SingleRollupWriteContext rollupWriteContext); synchronized void drainBatch();  }