docs: 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
      );
    }