private CompletableFuture < WriterFlushResult > mergeWith ( UpdateableSegmentMetadata transactionMetadata , MergeSegmentOperation mergeOp , TimeoutTimer timer ) { long traceId = LoggerHelpers . traceEnterWithContext ( log , this . traceObjectId , "mergeWith" , transactionMetadata . getId ( ) , transactionMetadata . getName ( ) , transactionMetadata . isSealedInStorage ( ) ) ; boolean emptySourceSegment = transactionMetadata . getLength ( ) == 0 ; if ( transactionMetadata . isDeleted ( ) && ! emptySourceSegment ) { // We came across a deleted source segment that had some data. We need to begin a reconciliation to figure out // the actual state of the segments in Storage. setState ( AggregatorState . ReconciliationNeeded ) ; return Futures . failedFuture ( new StreamSegmentNotExistsException ( transactionMetadata . getName ( ) ) ) ; } WriterFlushResult result = new WriterFlushResult ( ) ; CompletableFuture < SegmentProperties > merge ; if ( emptySourceSegment ) { // We came across a deleted source segment which had no data. No point in attempting to do anything, as any // operation involving this segment will complain about it not being there. log . warn ( "{}: Not applying '{}' because source segment is missing or empty." , this . traceObjectId , mergeOp ) ; merge = CompletableFuture . completedFuture ( this . metadata ) ; } else if ( ! transactionMetadata . isSealedInStorage ( ) || transactionMetadata . getLength ( ) > transactionMetadata . getStorageLength ( ) ) { // Nothing to do. Given Transaction is not eligible for merger yet. LoggerHelpers . traceLeave ( log , this . traceObjectId , "mergeWith" , traceId , result ) ; return CompletableFuture . completedFuture ( result ) ; } else { merge = mergeInStorage ( transactionMetadata , mergeOp , timer ) ; } return merge . thenAcceptAsync ( segmentProperties -> mergeCompleted ( segmentProperties , transactionMetadata , mergeOp ) , this . executor ) . thenComposeAsync ( v -> this . dataSource . deleteAllAttributes ( transactionMetadata , timer . getRemaining ( ) ) , this . executor ) . thenApply ( v -> { this . lastFlush . set ( this . timer . getElapsed ( ) ) ; result . withMergedBytes ( mergeOp . getLength ( ) ) ; LoggerHelpers . traceLeave ( log , this . traceObjectId , "mergeWith" , traceId , result ) ; return result ; } ) . exceptionally ( ex -> { Throwable realEx = Exceptions . unwrap ( ex ) ; if ( realEx instanceof BadOffsetException || realEx instanceof StreamSegmentNotExistsException ) { // We either attempted to write at an offset that already contained other data or the Transaction // Segment no longer exists. 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, and hope for the best. setState ( AggregatorState . ReconciliationNeeded ) ; } // Rethrow all exceptions. throw new CompletionException ( ex ) ; } ) ; }
Merges the Transaction StreamSegment with given metadata into this one at the current offset .