private CompletableFuture < WriterFlushResult > reconcileMergeOperation ( MergeSegmentOperation op , SegmentProperties storageInfo , TimeoutTimer timer ) { // Verify that the transaction segment is still registered in metadata. UpdateableSegmentMetadata transactionMeta = this . dataSource . getStreamSegmentMetadata ( op . getSourceSegmentId ( ) ) ; if ( transactionMeta == null ) { return Futures . failedFuture ( new ReconciliationFailureException ( String . format ( "Cannot reconcile operation '%s' because the source segment is missing from the metadata." , op ) , this . metadata , storageInfo ) ) ; } // Verify that the operation fits fully within this segment (mergers are atomic - they either merge all or nothing). if ( op . getLastStreamSegmentOffset ( ) > storageInfo . getLength ( ) ) { return Futures . failedFuture ( new ReconciliationFailureException ( String . format ( "Cannot reconcile operation '%s' because the source segment is not fully merged into the target." , op ) , this . metadata , storageInfo ) ) ; } // Verify that the transaction segment does not exist in Storage anymore. return this . storage . exists ( transactionMeta . getName ( ) , timer . getRemaining ( ) ) . thenComposeAsync ( exists -> { if ( exists ) { return Futures . failedFuture ( new ReconciliationFailureException ( String . format ( "Cannot reconcile operation '%s' because the transaction segment still exists in Storage." , op ) , this . metadata , storageInfo ) ) ; } // Clear out any attributes. return this . dataSource . deleteAllAttributes ( transactionMeta , timer . getRemaining ( ) ) ; } , this . executor ) . thenApplyAsync ( v -> { // Reconciliation complete. Pop the first operation off the list and update the metadata for the transaction segment. StorageOperation processedOperation = this . operations . removeFirst ( ) ; assert processedOperation != null && processedOperation instanceof MergeSegmentOperation : "First outstanding operation was not a MergeSegmentOperation" ; int newCount = this . mergeTransactionCount . decrementAndGet ( ) ; assert newCount >= 0 : "Negative value for mergeTransactionCount" ; // Since the operation is already reconciled, the StorageLength of this Segment must be at least // the last offset of the operation. We are about to invoke ReadIndex.completeMerge(), which requires // that this value be set to at least the last offset of the merged Segment, so we need to ensure it's // set now. This will also be set at the end of reconciliation, but we cannot wait until then to invoke // the callbacks. long minStorageLength = processedOperation . getLastStreamSegmentOffset ( ) ; if ( this . metadata . getStorageLength ( ) < minStorageLength ) { this . metadata . setStorageLength ( minStorageLength ) ; } updateMetadataForTransactionPostMerger ( transactionMeta , processedOperation . getStreamSegmentId ( ) ) ; return new WriterFlushResult ( ) . withMergedBytes ( op . getLength ( ) ) ; } , this . executor ) ; }
Attempts to reconcile the given MergeSegmentOperation .