@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()");
    }
  }
@ 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 ( ) ; 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()" ) ; } }
Unannounce the segments and wait for outstanding persists to finish.
Do not unlock base persist dir as we are not waiting for push executor to shut down
relying on current JVM to shutdown to not cause any locking problem if the task is restored.
In case when task is restored and current task is still active because of push executor (which it shouldn't be
since push executor starts daemon threads) then the locking should fail and new task should fail to start.
This also means that this method should only be called when task is shutting down.