ListenableFuture<SegmentsAndMetadata> pushInBackground(
      @Nullable final WrappedCommitter wrappedCommitter,
      final Collection<SegmentIdWithShardSpec> segmentIdentifiers,
      final boolean useUniquePath
  )
  {
    log.info("Pushing segments in background: [%s]", Joiner.on(", ").join(segmentIdentifiers));

    return Futures.transform(
        appenderator.push(segmentIdentifiers, wrappedCommitter, useUniquePath),
        (Function<SegmentsAndMetadata, SegmentsAndMetadata>) segmentsAndMetadata -> {
          // Sanity check
          final Set<SegmentIdWithShardSpec> pushedSegments = segmentsAndMetadata.getSegments().stream()
                                                                                .map(
                                                                                    SegmentIdWithShardSpec::fromDataSegment)
                                                                                .collect(Collectors.toSet());
          if (!pushedSegments.equals(Sets.newHashSet(segmentIdentifiers))) {
            log.warn(
                "Removing segments from deep storage because sanity check failed: %s", segmentsAndMetadata.getSegments()
            );

            segmentsAndMetadata.getSegments().forEach(dataSegmentKiller::killQuietly);

            throw new ISE(
                "WTF?! Pushed different segments than requested. Pushed[%s], requested[%s].",
                pushedSegments,
                segmentIdentifiers
            );
          }

          return segmentsAndMetadata;
        },
        executor
    );
  }
ListenableFuture < SegmentsAndMetadata > pushInBackground ( @ Nullable final WrappedCommitter wrappedCommitter , final Collection < SegmentIdWithShardSpec > segmentIdentifiers , final boolean useUniquePath ) { log . info ( "Pushing segments in background: [%s]" , Joiner . on ( ", " ) . join ( segmentIdentifiers ) ) ; return Futures . transform ( appenderator . push ( segmentIdentifiers , wrappedCommitter , useUniquePath ) , ( Function < SegmentsAndMetadata , SegmentsAndMetadata > ) segmentsAndMetadata -> { final Set < SegmentIdWithShardSpec > pushedSegments = segmentsAndMetadata . getSegments ( ) . stream ( ) . map ( SegmentIdWithShardSpec :: fromDataSegment ) . collect ( Collectors . toSet ( ) ) ; if ( ! pushedSegments . equals ( Sets . newHashSet ( segmentIdentifiers ) ) ) { log . warn ( "Removing segments from deep storage because sanity check failed: %s" , segmentsAndMetadata . getSegments ( ) ) ; segmentsAndMetadata . getSegments ( ) . forEach ( dataSegmentKiller :: killQuietly ) ; throw new ISE ( "WTF?! Pushed different segments than requested. Pushed[%s], requested[%s]." , pushedSegments , segmentIdentifiers ) ; } return segmentsAndMetadata ; } , executor ) ; }
Push the given segments in background.

@param wrappedCommitter   should not be null if you want to persist intermediate states
@param segmentIdentifiers identifiers of the segments to be pushed
@param useUniquePath      true if the segment should be written to a path with a unique identifier

@return a future for pushing segments