private Future < ? > runBatch ( Batch < T > batch ) throws InterruptedException { while ( true ) { try { runBatchMeter . mark ( ) ; if ( perRollupMetrics ) { batch . getMap ( ) . forEach ( ( addr , op ) -> { if ( addr . getSourceRollup ( ) . isPresent ( ) ) { updateRollupHistogram ( rollupWriteSize , addr . getSourceRollup ( ) . get ( ) , "rollupWriteSize" , op ) ; } } ) ; } return db . runBatchAsync ( batch , flushErrorHandler ) ; } catch ( FullQueueException e ) { asyncQueueBackoffMeter . mark ( ) ; // Sleeping and retrying like this means batches may be flushed out of order log . debug ( "Async queue is full, retrying soon" ) ; Thread . sleep ( 100 ) ; } } }
Hand off a batch to the DbHarness layer retrying on FullQueueException .