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);
    }
  }
private DataSegment mergeAndPush ( final SegmentIdWithShardSpec identifier , final Sink sink , final boolean useUniquePath ) { if ( sinks . get ( identifier ) != sink ) { log . warn ( "Sink for segment[%s] no longer valid, bailing out of mergeAndPush." , identifier ) ; return null ; } final File persistDir = computePersistDir ( identifier ) ; final File mergedTarget = new File ( persistDir , "merged" ) ; final File descriptorFile = computeDescriptorFile ( identifier ) ; 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 ( ) ) { if ( useUniquePath ) { 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 ( ) ; } final DataSegment segment = RetryUtils . retry ( ( ) -> 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 ) ; } }
Merge segment, push to deep storage. Should only be used on segments that have been fully persisted. Must only
be run in the single-threaded pushExecutor.

@param identifier    sink identifier
@param sink          sink to push
@param useUniquePath true if the segment should be written to a path with a unique identifier

@return segment descriptor, or null if the sink is no longer valid