private CompletableFuture < WriterFlushResult > flushPendingAppends ( Duration timeout ) { // Gather an InputStream made up of all the operations we can flush. FlushArgs flushArgs ; try { flushArgs = getFlushArgs ( ) ; } catch ( DataCorruptionException ex ) { return Futures . failedFuture ( ex ) ; } long traceId = LoggerHelpers . traceEnterWithContext ( log , this . traceObjectId , "flushPendingAppends" ) ; if ( flushArgs . getLength ( ) == 0 && flushArgs . getAttributes ( ) . isEmpty ( ) ) { // Nothing to flush. WriterFlushResult result = new WriterFlushResult ( ) ; LoggerHelpers . traceLeave ( log , this . traceObjectId , "flushPendingAppends" , traceId , result ) ; return CompletableFuture . completedFuture ( result ) ; } // Flush them. TimeoutTimer timer = new TimeoutTimer ( timeout ) ; CompletableFuture < Void > flush ; if ( flushArgs . getLength ( ) == 0 ) { flush = CompletableFuture . completedFuture ( null ) ; } else { flush = createSegmentIfNecessary ( ( ) -> this . storage . write ( this . handle . get ( ) , this . metadata . getStorageLength ( ) , flushArgs . getStream ( ) , flushArgs . getLength ( ) , timer . getRemaining ( ) ) , timer . getRemaining ( ) ) ; } if ( ! flushArgs . getAttributes ( ) . isEmpty ( ) ) { flush = flush . thenComposeAsync ( v -> handleAttributeException ( this . dataSource . persistAttributes ( this . metadata . getId ( ) , flushArgs . attributes , timer . getRemaining ( ) ) ) ) ; } return flush . thenApplyAsync ( v -> { WriterFlushResult result = updateStatePostFlush ( flushArgs ) ; LoggerHelpers . traceLeave ( log , this . traceObjectId , "flushPendingAppends" , traceId , result ) ; return result ; } , this . executor ) . exceptionally ( ex -> { if ( Exceptions . unwrap ( ex ) instanceof BadOffsetException ) { // We attempted to write at an offset that already contained other data. This can happen for a number of // reasons, but we do not have enough information here to determine why. We need to enter reconciliation // mode, which will determine the actual state of the segment in storage and take appropriate actions. setState ( AggregatorState . ReconciliationNeeded ) ; } // Rethrow all exceptions. throw new CompletionException ( ex ) ; } ) ; }
Flushes all Append Operations that can be flushed up to the maximum allowed flush size .