salesforce-lib/src/main/java/com/streamsets/pipeline/stage/origin/salesforce/ForceSource.java
@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() {
    // 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();
  }
{@inheritDoc}