private CompletableFuture < Void > compactIfNeeded ( DirectSegmentAccess segment , long highestCopiedOffset , TimeoutTimer timer ) { // Decide if compaction is needed. If not, bail out early. SegmentProperties info = segment . getInfo ( ) ; CompletableFuture < Void > result ; if ( this . compactor . isCompactionRequired ( info ) ) { result = this . compactor . compact ( segment , timer ) ; } else { log . debug ( "{}: No compaction required at this time." , this . traceObjectId ) ; result = CompletableFuture . completedFuture ( null ) ; } return result . thenComposeAsync ( v -> { // Calculate the safe truncation offset. long truncateOffset = this . compactor . calculateTruncationOffset ( segment . getInfo ( ) , highestCopiedOffset ) ; // Truncate if necessary. if ( truncateOffset > 0 ) { log . debug ( "{}: Truncating segment at offset {}." , this . traceObjectId , truncateOffset ) ; return segment . truncate ( truncateOffset , timer . getRemaining ( ) ) ; } else { log . debug ( "{}: No segment truncation possible now." , this . traceObjectId ) ; return CompletableFuture . completedFuture ( null ) ; } } , this . executor ) . exceptionally ( ex -> { // We want to record the compaction failure, but since this is not a critical step in making progress, // we do not want to prevent the StorageWriter from ack-ing operations. log . error ( "{}: Compaction failed." , this . traceObjectId , ex ) ; return null ; } ) ; }
Performs a Table Segment Compaction if needed .