docs: public ListenableFuture<?> closeAndDelete() {
  
          // We're going away, so shut ourselves from the external world
          m_closed = true;
          m_ackMailboxRefs.set(null);
  
          // Export mastership should have been released: force it.
          m_mastershipAccepted.set(false);
  
          // FIXME: necessary? Old processor should have been shut down.
          // Returning null indicates end of stream
          try {
              if (m_pollTask != null) {
                  m_pollTask.setFuture(null);
              }
          } catch (RejectedExecutionException reex) {
              // Ignore, {@code GuestProcessor} was closed
          }
          m_pollTask = null;
  
          return m_es.submit(new Runnable() {
              @Override
              public void run() {
                  try {
                      // Discard the pending container, shortcutting the standard discard logic
                      AckingContainer ack = m_pendingContainer.getAndSet(null);
                      if (ack != null) {
                          if (exportLog.isDebugEnabled()) {
                              exportLog.debug("Discard pending container, lastSeqNo: " + ack.getLastSeqNo());
                          }
                          ack.internalDiscard();
                      }
                      m_committedBuffers.closeAndDelete();
                      m_adFile.delete();
                  } catch(IOException e) {
                      exportLog.rateLimitedLog(60, Level.WARN, e, "Error closing commit buffers");
                  } finally {
                      m_es.shutdown();
                  }
              }
          });
      }