private void completeFutures ( ) { long commitIndex = queues . values ( ) . stream ( ) . map ( queue -> queue . ackedIndex ) . reduce ( Math :: min ) . orElse ( 0L ) ; for ( long i = context . getCommitIndex ( ) + 1 ; i <= commitIndex ; i ++ ) { CompletableFuture < Void > future = futures . remove ( i ) ; if ( future != null ) { future . complete ( null ) ; } } context . setCommitIndex ( commitIndex ) ; }
Completes futures .