void completeMerge ( SegmentMetadata sourceMetadata ) { long traceId = LoggerHelpers . traceEnterWithContext ( log , this . traceObjectId , "completeMerge" , sourceMetadata . getId ( ) ) ; Exceptions . checkNotClosed ( this . closed , this ) ; Exceptions . checkArgument ( sourceMetadata . isDeleted ( ) , "sourceSegmentStreamId" , "Given StreamSegmentReadIndex refers to a StreamSegment that has not been deleted yet." ) ; if ( sourceMetadata . getLength ( ) == 0 ) { // beginMerge() does not take any action when the source Segment is empty, so there is nothing for us to do either. return ; } // Find the appropriate redirect entry. RedirectIndexEntry redirectEntry ; PendingMerge pendingMerge ; synchronized ( this . lock ) { pendingMerge = this . pendingMergers . getOrDefault ( sourceMetadata . getId ( ) , null ) ; Exceptions . checkArgument ( pendingMerge != null , "sourceSegmentStreamId" , "Given StreamSegmentReadIndex's merger with this one has not been initiated using beginMerge. Cannot finalize the merger." ) ; // Get the RedirectIndexEntry. These types of entries are sticky in the cache and DO NOT contribute to the // cache Stats. They are already accounted for in the other Segment's ReadIndex. ReadIndexEntry indexEntry = this . indexEntries . get ( pendingMerge . getMergeOffset ( ) ) ; assert indexEntry != null && ! indexEntry . isDataEntry ( ) : String . format ( "pendingMergers points to a ReadIndexEntry that does not exist or is of the wrong type. sourceStreamSegmentId = %d, offset = %d, treeEntry = %s." , sourceMetadata . getId ( ) , pendingMerge . getMergeOffset ( ) , indexEntry ) ; redirectEntry = ( RedirectIndexEntry ) indexEntry ; } StreamSegmentReadIndex sourceIndex = redirectEntry . getRedirectReadIndex ( ) ; // Get all the entries from the source index and append them here. List < MergedIndexEntry > sourceEntries = sourceIndex . getAllEntries ( redirectEntry . getStreamSegmentOffset ( ) ) ; synchronized ( this . lock ) { // Remove redirect entry (again, no need to update the Cache Stats, as this is a RedirectIndexEntry). this . indexEntries . remove ( pendingMerge . getMergeOffset ( ) ) ; this . pendingMergers . remove ( sourceMetadata . getId ( ) ) ; sourceEntries . forEach ( this :: addToIndex ) ; } List < FutureReadResultEntry > pendingReads = pendingMerge . seal ( ) ; if ( pendingReads . size ( ) > 0 ) { log . debug ( "{}: triggerFutureReads for Pending Merge (Count = {}, MergeOffset = {}, MergeLength = {})." , this . traceObjectId , pendingReads . size ( ) , pendingMerge . getMergeOffset ( ) , sourceIndex . getSegmentLength ( ) ) ; triggerFutureReads ( pendingReads ) ; } LoggerHelpers . traceLeave ( log , this . traceObjectId , "completeMerge" , traceId ) ; }
Executes Step 2 of the 2 - Step Merge Process . The StreamSegments are physically merged in the Storage . The Source StreamSegment does not exist anymore . The ReadIndex entries of the two Streams are actually joined together .