code: 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);
      }
docs: Executes the merge of the Source StreamSegment with given metadata into this one in Storage.