@Override
  public void destroy() {
    // SDC-6258 - destroy() is called from a different thread, so we
    // need to signal produce() to terminate early
    destroyed.set(true);

    if (job != null) {
      try {
        try {
          bulkConnection.abortJob(job.getId());
        } catch (AsyncApiException e) {
          ForceUtils.renewSession(bulkConnection, e);
          bulkConnection.abortJob(job.getId());
        }
        job = null;
      } catch (AsyncApiException e) {
        LOG.error("Exception while aborting job", e);
      }
    }

    job = null;

    if (forceConsumer != null) {
      try {
        forceConsumer.stop();
        forceConsumer = null;
      } catch (Exception e) {
        LOG.error("Exception while stopping ForceStreamConsumer.", e);
      }

      if (!messageQueue.isEmpty()) {
        LOG.error("Queue still had {} entities at shutdown.", messageQueue.size());
      } else {
        LOG.info("Queue was empty at shutdown. No data lost.");
      }
    }

    if (recordCreator != null) {
      recordCreator.destroy();
    }

    // Clean up any open resources.
    super.destroy();
  }
@ Override public void destroy ( ) { destroyed . set ( true ) ; if ( job != null ) { try { try { bulkConnection . abortJob ( job . getId ( ) ) ; } catch ( AsyncApiException e ) { ForceUtils . renewSession ( bulkConnection , e ) ; bulkConnection . abortJob ( job . getId ( ) ) ; } job = null ; } catch ( AsyncApiException e ) { LOG . error ( "Exception while aborting job" , e ) ; } } job = null ; if ( forceConsumer != null ) { try { forceConsumer . stop ( ) ; forceConsumer = null ; } catch ( Exception e ) { LOG . error ( "Exception while stopping ForceStreamConsumer." , e ) ; } if ( ! messageQueue . isEmpty ( ) ) { LOG . error ( "Queue still had {} entities at shutdown." , messageQueue . size ( ) ) ; } else { LOG . info ( "Queue was empty at shutdown. No data lost." ) ; } } if ( recordCreator != null ) { recordCreator . destroy ( ) ; } super . destroy ( ) ; }
{@inheritDoc}