docs: private DataSegment mergeAndPush(final SegmentIdWithShardSpec identifier, final Sink sink, final boolean useUniquePath)
    {
      // Bail out if this sink is null or otherwise not what we expect.
      if (sinks.get(identifier) != sink) {
        log.warn("Sink for segment[%s] no longer valid, bailing out of mergeAndPush.", identifier);
        return null;
      }
  
      // Use a descriptor file to indicate that pushing has completed.
      final File persistDir = computePersistDir(identifier);
      final File mergedTarget = new File(persistDir, "merged");
      final File descriptorFile = computeDescriptorFile(identifier);
  
      // Sanity checks
      for (FireHydrant hydrant : sink) {
        if (sink.isWritable()) {
          throw new ISE("WTF?! Expected sink to be no longer writable before mergeAndPush. Segment[%s].", identifier);
        }
  
        synchronized (hydrant) {
          if (!hydrant.hasSwapped()) {
            throw new ISE("WTF?! Expected sink to be fully persisted before mergeAndPush. Segment[%s].", identifier);
          }
        }
      }
  
      try {
        if (descriptorFile.exists()) {
          // Already pushed.
  
          if (useUniquePath) {
            // Don't reuse the descriptor, because the caller asked for a unique path. Leave the old one as-is, since
            // it might serve some unknown purpose.
            log.info("Pushing segment[%s] again with new unique path.", identifier);
          } else {
            log.info("Segment[%s] already pushed.", identifier);
            return objectMapper.readValue(descriptorFile, DataSegment.class);
          }
        }
  
        log.info("Pushing merged index for segment[%s].", identifier);
  
        removeDirectory(mergedTarget);
  
        if (mergedTarget.exists()) {
          throw new ISE("Merged target[%s] exists after removing?!", mergedTarget);
        }
  
        final File mergedFile;
        List<QueryableIndex> indexes = new ArrayList<>();
        Closer closer = Closer.create();
        try {
          for (FireHydrant fireHydrant : sink) {
            Pair<Segment, Closeable> segmentAndCloseable = fireHydrant.getAndIncrementSegment();
            final QueryableIndex queryableIndex = segmentAndCloseable.lhs.asQueryableIndex();
            log.info("Adding hydrant[%s]", fireHydrant);
            indexes.add(queryableIndex);
            closer.register(segmentAndCloseable.rhs);
          }
  
          mergedFile = indexMerger.mergeQueryableIndex(
              indexes,
              schema.getGranularitySpec().isRollup(),
              schema.getAggregators(),
              mergedTarget,
              tuningConfig.getIndexSpec(),
              tuningConfig.getSegmentWriteOutMediumFactory()
          );
        }
        catch (Throwable t) {
          throw closer.rethrow(t);
        }
        finally {
          closer.close();
        }
  
        // Retry pushing segments because uploading to deep storage might fail especially for cloud storage types
        final DataSegment segment = RetryUtils.retry(
            // The appenderator is currently being used for the local indexing task and the Kafka indexing task. For the
            // Kafka indexing task, pushers must use unique file paths in deep storage in order to maintain exactly-once
            // semantics.
            () -> dataSegmentPusher.push(
                mergedFile,
                sink.getSegment().withDimensions(IndexMerger.getMergedDimensionsFromQueryableIndexes(indexes)),
                useUniquePath
            ),
            exception -> exception instanceof Exception,
            5
        );
  
        objectMapper.writeValue(descriptorFile, segment);
  
        log.info("Pushed merged index for segment[%s], descriptor is: %s", identifier, segment);
  
        return segment;
      }
      catch (Exception e) {
        metrics.incrementFailedHandoffs();
        log.warn(e, "Failed to push merged index for segment[%s].", identifier);
        throw new RuntimeException(e);
      }
    }