private CompletableFuture < SegmentProperties > mergeInStorage ( SegmentMetadata transactionMetadata , MergeSegmentOperation mergeOp , TimeoutTimer timer ) { return this . storage . getStreamSegmentInfo ( transactionMetadata . getName ( ) , timer . getRemaining ( ) ) . thenAcceptAsync ( transProperties -> { // One last verification before the actual merger: // Check that the Storage agrees with our metadata (if not, we have a problem ...) if ( transProperties . getLength ( ) != transactionMetadata . getStorageLength ( ) ) { throw new CompletionException ( new DataCorruptionException ( String . format ( "Transaction Segment '%s' cannot be merged into parent '%s' because its metadata disagrees with the Storage. Metadata.StorageLength=%d, Storage.StorageLength=%d" , transactionMetadata . getName ( ) , this . metadata . getName ( ) , transactionMetadata . getStorageLength ( ) , transProperties . getLength ( ) ) ) ) ; } if ( transProperties . getLength ( ) != mergeOp . getLength ( ) ) { throw new CompletionException ( new DataCorruptionException ( String . format ( "Transaction Segment '%s' cannot be merged into parent '%s' because the declared length in the operation disagrees with the Storage. Operation.Length=%d, Storage.StorageLength=%d" , transactionMetadata . getName ( ) , this . metadata . getName ( ) , mergeOp . getLength ( ) , transProperties . getLength ( ) ) ) ) ; } } , this . executor ) . thenComposeAsync ( v -> createSegmentIfNecessary ( ( ) -> storage . concat ( this . handle . get ( ) , mergeOp . getStreamSegmentOffset ( ) , transactionMetadata . getName ( ) , timer . getRemaining ( ) ) , timer . getRemaining ( ) ) , this . executor ) . exceptionally ( ex -> { ex = Exceptions . unwrap ( ex ) ; if ( transactionMetadata . getLength ( ) == 0 && ex instanceof StreamSegmentNotExistsException && ( ( StreamSegmentNotExistsException ) ex ) . getStreamSegmentName ( ) . equals ( transactionMetadata . getName ( ) ) ) { log . warn ( "{}: Not applying '{}' because source segment is missing (storage) and had no data." , this . traceObjectId , mergeOp ) ; return null ; } else { throw new CompletionException ( ex ) ; } } ) . thenComposeAsync ( v -> storage . getStreamSegmentInfo ( this . metadata . getName ( ) , timer . getRemaining ( ) ) , this . executor ) ; }
Executes the merge of the Source StreamSegment with given metadata into this one in Storage .