docs: public ListenableFuture<SegmentsAndMetadata> registerHandoff(SegmentsAndMetadata segmentsAndMetadata)
    {
      if (segmentsAndMetadata == null) {
        return Futures.immediateFuture(null);
  
      } else {
        final List<SegmentIdWithShardSpec> waitingSegmentIdList = segmentsAndMetadata.getSegments().stream()
                                                                                     .map(
                                                                                         SegmentIdWithShardSpec::fromDataSegment)
                                                                                     .collect(Collectors.toList());
        final Object metadata = Preconditions.checkNotNull(segmentsAndMetadata.getCommitMetadata(), "commitMetadata");
  
        if (waitingSegmentIdList.isEmpty()) {
          return Futures.immediateFuture(
              new SegmentsAndMetadata(
                  segmentsAndMetadata.getSegments(),
                  ((AppenderatorDriverMetadata) metadata).getCallerMetadata()
              )
          );
        }
  
        log.info("Register handoff of segments: [%s]", waitingSegmentIdList);
  
        final SettableFuture<SegmentsAndMetadata> resultFuture = SettableFuture.create();
        final AtomicInteger numRemainingHandoffSegments = new AtomicInteger(waitingSegmentIdList.size());
  
        for (final SegmentIdWithShardSpec segmentIdentifier : waitingSegmentIdList) {
          handoffNotifier.registerSegmentHandoffCallback(
              new SegmentDescriptor(
                  segmentIdentifier.getInterval(),
                  segmentIdentifier.getVersion(),
                  segmentIdentifier.getShardSpec().getPartitionNum()
              ),
              Execs.directExecutor(),
              () -> {
                log.info("Segment[%s] successfully handed off, dropping.", segmentIdentifier);
                metrics.incrementHandOffCount();
  
                final ListenableFuture<?> dropFuture = appenderator.drop(segmentIdentifier);
                Futures.addCallback(
                    dropFuture,
                    new FutureCallback<Object>()
                    {
                      @Override
                      public void onSuccess(Object result)
                      {
                        if (numRemainingHandoffSegments.decrementAndGet() == 0) {
                          log.info("Successfully handed off [%d] segments.", segmentsAndMetadata.getSegments().size());
                          resultFuture.set(
                              new SegmentsAndMetadata(
                                  segmentsAndMetadata.getSegments(),
                                  ((AppenderatorDriverMetadata) metadata).getCallerMetadata()
                              )
                          );
                        }
                      }
  
                      @Override
                      public void onFailure(Throwable e)
                      {
                        log.warn(e, "Failed to drop segment[%s]?!", segmentIdentifier);
                        numRemainingHandoffSegments.decrementAndGet();
                        resultFuture.setException(e);
                      }
                    }
                );
              }
          );
        }
  
        return resultFuture;
      }
    }