docs: @Override
    public void closeNow()
    {
      if (!closed.compareAndSet(false, true)) {
        log.info("Appenderator already closed");
        return;
      }
  
      log.info("Shutting down immediately...");
      for (Map.Entry<SegmentIdWithShardSpec, Sink> entry : sinks.entrySet()) {
        try {
          segmentAnnouncer.unannounceSegment(entry.getValue().getSegment());
        }
        catch (Exception e) {
          log.makeAlert(e, "Failed to unannounce segment[%s]", schema.getDataSource())
             .addData("identifier", entry.getKey().toString())
             .emit();
        }
      }
      try {
        shutdownExecutors();
        // We don't wait for pushExecutor to be terminated. See Javadoc for more details.
        Preconditions.checkState(
            persistExecutor == null || persistExecutor.awaitTermination(365, TimeUnit.DAYS),
            "persistExecutor not terminated"
        );
        Preconditions.checkState(
            intermediateTempExecutor == null || intermediateTempExecutor.awaitTermination(365, TimeUnit.DAYS),
            "intermediateTempExecutor not terminated"
        );
        persistExecutor = null;
        intermediateTempExecutor = null;
      }
      catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new ISE("Failed to shutdown executors during close()");
      }
    }