docs: private void persistAndMerge(final long truncatedTime, final Sink sink)
    {
      final String threadName = StringUtils.format(
          "%s-%s-persist-n-merge", schema.getDataSource(), DateTimes.utc(truncatedTime)
      );
      mergeExecutor.execute(
          new ThreadRenamingRunnable(threadName)
          {
            final Interval interval = sink.getInterval();
            Stopwatch mergeStopwatch = null;
  
            @Override
            public void doRun()
            {
              try {
                // Bail out if this sink has been abandoned by a previously-executed task.
                if (sinks.get(truncatedTime) != sink) {
                  log.info("Sink[%s] was abandoned, bailing out of persist-n-merge.", sink);
                  return;
                }
  
                // Use a file to indicate that pushing has completed.
                final File persistDir = computePersistDir(schema, interval);
                final File mergedTarget = new File(persistDir, "merged");
                final File isPushedMarker = new File(persistDir, "isPushedMarker");
  
                if (!isPushedMarker.exists()) {
                  removeSegment(sink, mergedTarget);
                  if (mergedTarget.exists()) {
                    log.wtf("Merged target[%s] exists?!", mergedTarget);
                    return;
                  }
                } else {
                  log.info("Already pushed sink[%s]", sink);
                  return;
                }
  
              /*
              Note: it the plumber crashes after persisting a subset of hydrants then might duplicate data as these
              hydrants will be read but older commitMetadata will be used. fixing this possibly needs structural
              changes to plumber.
               */
                for (FireHydrant hydrant : sink) {
                  synchronized (hydrant) {
                    if (!hydrant.hasSwapped()) {
                      log.info("Hydrant[%s] hasn't swapped yet, swapping. Sink[%s]", hydrant, sink);
                      final int rowCount = persistHydrant(hydrant, schema, interval, null);
                      metrics.incrementRowOutputCount(rowCount);
                    }
                  }
                }
                final long mergeThreadCpuTime = JvmUtils.safeGetThreadCpuTime();
                mergeStopwatch = Stopwatch.createStarted();
  
                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,
                      config.getIndexSpec(),
                      config.getSegmentWriteOutMediumFactory()
                  );
                }
                catch (Throwable t) {
                  throw closer.rethrow(t);
                }
                finally {
                  closer.close();
                }
  
                // emit merge metrics before publishing segment
                metrics.incrementMergeCpuTime(JvmUtils.safeGetThreadCpuTime() - mergeThreadCpuTime);
                metrics.incrementMergeTimeMillis(mergeStopwatch.elapsed(TimeUnit.MILLISECONDS));
  
                log.info("Pushing [%s] to deep storage", sink.getSegment().getId());
  
                DataSegment segment = dataSegmentPusher.push(
                    mergedFile,
                    sink.getSegment().withDimensions(IndexMerger.getMergedDimensionsFromQueryableIndexes(indexes)),
                    false
                );
                log.info("Inserting [%s] to the metadata store", sink.getSegment().getId());
                segmentPublisher.publishSegment(segment);
  
                if (!isPushedMarker.createNewFile()) {
                  log.makeAlert("Failed to create marker file for [%s]", schema.getDataSource())
                     .addData("interval", sink.getInterval())
                     .addData("partitionNum", segment.getShardSpec().getPartitionNum())
                     .addData("marker", isPushedMarker)
                     .emit();
                }
              }
              catch (Exception e) {
                metrics.incrementFailedHandoffs();
                log.makeAlert(e, "Failed to persist merged index[%s]", schema.getDataSource())
                   .addData("interval", interval)
                   .emit();
                if (shuttingDown) {
                  // We're trying to shut down, and this segment failed to push. Let's just get rid of it.
                  // This call will also delete possibly-partially-written files, so we don't need to do it explicitly.
                  cleanShutdown = false;
                  abandonSegment(truncatedTime, sink);
                }
              }
              finally {
                if (mergeStopwatch != null) {
                  mergeStopwatch.stop();
                }
              }
            }
          }
      );
      handoffNotifier.registerSegmentHandoffCallback(
          new SegmentDescriptor(sink.getInterval(), sink.getVersion(), config.getShardSpec().getPartitionNum()),
          mergeExecutor, new Runnable()
          {
            @Override
            public void run()
            {
              abandonSegment(sink.getInterval().getStartMillis(), sink);
              metrics.incrementHandOffCount();
            }
          }
      );
    }