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);
    }
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 ) { return ; } 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." ) ; 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 ( ) ; List < MergedIndexEntry > sourceEntries = sourceIndex . getAllEntries ( redirectEntry . getStreamSegmentOffset ( ) ) ; synchronized ( this . lock ) { 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.

@param sourceMetadata The SegmentMetadata of the Segment that was merged into this one.