docs: ListenableFuture<SegmentsAndMetadata> publishInBackground(
        SegmentsAndMetadata segmentsAndMetadata,
        TransactionalSegmentPublisher publisher
    )
    {
      return executor.submit(
          () -> {
            if (segmentsAndMetadata.getSegments().isEmpty()) {
              log.info("Nothing to publish, skipping publish step.");
            } else {
              log.info(
                  "Publishing segments with commitMetadata[%s]: [%s]",
                  segmentsAndMetadata.getCommitMetadata(),
                  Joiner.on(", ").join(segmentsAndMetadata.getSegments())
              );
  
              try {
                final Object metadata = segmentsAndMetadata.getCommitMetadata();
                final ImmutableSet<DataSegment> ourSegments = ImmutableSet.copyOf(segmentsAndMetadata.getSegments());
                final SegmentPublishResult publishResult = publisher.publishSegments(
                    ourSegments,
                    metadata == null ? null : ((AppenderatorDriverMetadata) metadata).getCallerMetadata()
                );
  
                if (publishResult.isSuccess()) {
                  log.info("Published segments.");
                } else {
                  // Publishing didn't affirmatively succeed. However, segments with our identifiers may still be active
                  // now after all, for two possible reasons:
                  //
                  // 1) A replica may have beat us to publishing these segments. In this case we want to delete the
                  //    segments we pushed (if they had unique paths) to avoid wasting space on deep storage.
                  // 2) We may have actually succeeded, but not realized it due to missing the confirmation response
                  //    from the overlord. In this case we do not want to delete the segments we pushed, since they are
                  //    now live!
  
                  final Set<SegmentIdWithShardSpec> segmentsIdentifiers = segmentsAndMetadata
                      .getSegments()
                      .stream()
                      .map(SegmentIdWithShardSpec::fromDataSegment)
                      .collect(Collectors.toSet());
  
                  final Set<DataSegment> activeSegments = usedSegmentChecker.findUsedSegments(segmentsIdentifiers);
  
                  if (activeSegments.equals(ourSegments)) {
                    log.info("Could not publish segments, but checked and found them already published. Continuing.");
  
                    // Clean up pushed segments if they are physically disjoint from the published ones (this means
                    // they were probably pushed by a replica, and with the unique paths option).
                    final boolean physicallyDisjoint = Sets.intersection(
                        activeSegments.stream().map(DataSegment::getLoadSpec).collect(Collectors.toSet()),
                        ourSegments.stream().map(DataSegment::getLoadSpec).collect(Collectors.toSet())
                    ).isEmpty();
  
                    if (physicallyDisjoint) {
                      segmentsAndMetadata.getSegments().forEach(dataSegmentKiller::killQuietly);
                    }
                  } else {
                    // Our segments aren't active. Publish failed for some reason. Clean them up and then throw an error.
                    segmentsAndMetadata.getSegments().forEach(dataSegmentKiller::killQuietly);
  
                    if (publishResult.getErrorMsg() != null) {
                      throw new ISE("Failed to publish segments because of [%s].", publishResult.getErrorMsg());
                    } else {
                      throw new ISE("Failed to publish segments.");
                    }
                  }
                }
              }
              catch (Exception e) {
                // Must not remove segments here, we aren't sure if our transaction succeeded or not.
                log.warn(e, "Failed publish, not removing segments: %s", segmentsAndMetadata.getSegments());
                Throwables.propagateIfPossible(e);
                throw new RuntimeException(e);
              }
            }
  
            return segmentsAndMetadata;
          }
      );
    }