private CompletableFuture < WriterFlushResult > sealIfNecessary ( WriterFlushResult flushResult , TimeoutTimer timer ) { if ( ! this . hasSealPending . get ( ) || ! ( this . operations . getFirst ( ) instanceof StreamSegmentSealOperation ) ) { // Either no Seal is pending or the next operation is not a seal - we cannot execute a seal. return CompletableFuture . completedFuture ( flushResult ) ; } long traceId = LoggerHelpers . traceEnterWithContext ( log , this . traceObjectId , "sealIfNecessary" ) ; CompletableFuture < Void > sealTask ; if ( this . handle . get ( ) == null ) { // Segment is empty. It might not have been created yet, so don't do anything. assert this . metadata . getStorageLength ( ) == 0 : "handle is null but Metadata.StorageLength is non-zero" ; sealTask = CompletableFuture . completedFuture ( null ) ; } else { // Segment is non-empty; it should exist. sealTask = this . storage . seal ( this . handle . get ( ) , timer . getRemaining ( ) ) ; } return sealTask . thenComposeAsync ( v -> sealAttributes ( timer . getRemaining ( ) ) , this . executor ) . handleAsync ( ( v , ex ) -> { ex = Exceptions . unwrap ( ex ) ; if ( ex != null && ! ( ex instanceof StreamSegmentSealedException ) ) { // The operation failed, and it was not because the Segment was already Sealed. Throw it again. // We consider the Seal to succeed if the Segment in Storage is already sealed - it's an idempotent operation. if ( ex instanceof StreamSegmentNotExistsException ) { // The segment does not exist (anymore). This is possible if it got merged into another one. // Begin a reconciliation to determine the actual state. setState ( AggregatorState . ReconciliationNeeded ) ; } throw new CompletionException ( ex ) ; } updateStatePostSeal ( ) ; LoggerHelpers . traceLeave ( log , this . traceObjectId , "sealIfNecessary" , traceId , flushResult ) ; return flushResult ; } , this . executor ) ; }
Seals the StreamSegment in Storage if necessary .