@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;
                });
    }
@ Override public CompletableFuture < WriterFlushResult > flush ( Duration timeout ) { ensureInitializedAndNotClosed ( ) ; if ( this . metadata . isDeletedInStorage ( ) ) { 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 ; 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 ) { 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.

@param timeout Timeout for the operation.
@return A CompletableFuture that, when completed, will contain a summary of the flush operation. If any errors
occurred during the flush, the Future will be completed with the appropriate exception.