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 ) ; }
Push the given segments in background .