@ Override public CompletableFuture < WriterFlushResult > flush ( Duration timeout ) { ensureInitializedAndNotClosed ( ) ; if ( this . metadata . isDeletedInStorage ( ) ) { // Segment has been deleted; don't do anything else. return CompletableFuture . completedFuture ( new WriterFlushResult ( ) ) ; } long traceId = LoggerHelpers . traceEnterWithContext ( log , this . traceObjectId , "flush" ) ; TimeoutTimer timer = new TimeoutTimer ( timeout ) ; CompletableFuture < WriterFlushResult > result ; try { switch ( this . state . get ( ) ) { case Writing : result = flushNormally ( timer ) ; break ; case ReconciliationNeeded : result = beginReconciliation ( timer ) . thenComposeAsync ( v -> reconcile ( timer ) , this . executor ) ; break ; case Reconciling : result = reconcile ( timer ) ; break ; //$CASES-OMITTED$ default : result = Futures . failedFuture ( new IllegalStateException ( String . format ( "Unexpected state for SegmentAggregator (%s) for segment '%s'." , this . state , this . metadata . getName ( ) ) ) ) ; break ; } } catch ( Exception ex ) { // Convert synchronous errors into async errors - it's easier to handle on the receiving end. result = Futures . failedFuture ( ex ) ; } return result . thenApply ( r -> { LoggerHelpers . traceLeave ( log , this . traceObjectId , "flush" , traceId , r ) ; return r ; } ) ; }
Flushes the contents of the Aggregator to the Storage .