jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/multithread/JdbcBaseRunnable.java
protected void handlePostBatchAsNeeded(
      boolean resultSetEndReached,
      int recordCount,
      int eventCount,
      BatchContext batchContext
  ) {
    AtomicBoolean shouldEvict = new AtomicBoolean(resultSetEndReached);
    // If we read at least one record, it's safe to drop the starting offsets, otherwise they have to re-used
    if(recordCount > 0) {
      tableRuntimeContext.getSourceTableContext().clearStartOffset();
    }

    //Only process batch if there are records or events
    if (recordCount > 0 || eventCount > 0) {
      TableReadContext tableReadContext = tableReadContextCache.getIfPresent(tableRuntimeContext);
      Optional.ofNullable(tableReadContext)
          .ifPresent(readContext -> {
            readContext.addProcessingMetrics(1, recordCount);
            LOG.debug(
                "Table {} read batches={} and records={}",
                tableRuntimeContext.getQualifiedName(),
                readContext.getNumberOfBatches(),
                readContext.getNumberOfRecords()
            );
            calculateEvictTableFlag(shouldEvict, tableReadContext);
          });
      updateGauge(JdbcBaseRunnable.Status.BATCH_GENERATED);

      //Process And Commit offsets
      if (tableRuntimeContext.isUsingNonIncrementalLoad()) {
        // process the batch now, will handle the offset commit outside this block
        context.processBatch(batchContext);
      } else {
        // for incremental (normal) mode, the offset was already stored in this map
        // by the specific subclass's createAndAddRecord method
        final String offsetValue = offsets.get(tableRuntimeContext.getOffsetKey());
        context.processBatch(batchContext, tableRuntimeContext.getOffsetKey(), offsetValue);
      }

    }

    if (tableRuntimeContext.isUsingNonIncrementalLoad()) {
      // for non-incremental mode, the offset is simply a singleton map indicating whether it's finished
      final String offsetValue = createNonIncrementalLoadOffsetValue(resultSetEndReached);
      context.commitOffset(tableRuntimeContext.getOffsetKey(), offsetValue);
    }

    //Make sure we close the result set only when there are no more rows in the result set
    if (shouldEvict.get()) {
      //Invalidate so as to fetch a new result set
      //We close the result set/statement in Removal Listener
      tableReadContextCache.invalidate(tableRuntimeContext);

      tableProvider.releaseOwnedTable(tableRuntimeContext, threadNumber);

      tableRuntimeContext = null;
    } else if (
        tableJdbcConfigBean.batchTableStrategy == BatchTableStrategy.SWITCH_TABLES
            && !tableRuntimeContext.isUsingNonIncrementalLoad()) {
      tableRuntimeContext = null;
    }

    final List<TableRuntimeContext> removedPartitions = tableProvider.getAndClearRemovedPartitions();
    if (removedPartitions != null && removedPartitions.size() > 0) {
      for (TableRuntimeContext partition : removedPartitions) {
        LOG.debug(
            "Removing offset entry for partition {} since it has been removed from the table provider",
            partition.getDescription()
        );
        context.commitOffset(partition.getOffsetKey(), null);
      }
    }
  }
protected void handlePostBatchAsNeeded(
      boolean resultSetEndReached,
      int recordCount,
      int eventCount,
      BatchContext batchContext
  ) {
    AtomicBoolean shouldEvict = new AtomicBoolean(resultSetEndReached);
    // If we read at least one record, it's safe to drop the starting offsets, otherwise they have to re-used
    if(recordCount > 0) {
      tableRuntimeContext.getSourceTableContext().clearStartOffset();
    }

    //Only process batch if there are records or events
    if (recordCount > 0 || eventCount > 0) {
      TableReadContext tableReadContext = tableReadContextCache.getIfPresent(tableRuntimeContext);
      Optional.ofNullable(tableReadContext)
          .ifPresent(readContext -> {
            readContext.addProcessingMetrics(1, recordCount);
            LOG.debug(
                "Table {} read batches={} and records={}",
                tableRuntimeContext.getQualifiedName(),
                readContext.getNumberOfBatches(),
                readContext.getNumberOfRecords()
            );
            calculateEvictTableFlag(shouldEvict, tableReadContext);
          });
      updateGauge(JdbcBaseRunnable.Status.BATCH_GENERATED);

      //Process And Commit offsets
      if (tableRuntimeContext.isUsingNonIncrementalLoad()) {
        // process the batch now, will handle the offset commit outside this block
        context.processBatch(batchContext);
      } else {
        // for incremental (normal) mode, the offset was already stored in this map
        // by the specific subclass's createAndAddRecord method
        final String offsetValue = offsets.get(tableRuntimeContext.getOffsetKey());
        context.processBatch(batchContext, tableRuntimeContext.getOffsetKey(), offsetValue);
      }

    }

    if (tableRuntimeContext.isUsingNonIncrementalLoad()) {
      // for non-incremental mode, the offset is simply a singleton map indicating whether it's finished
      final String offsetValue = createNonIncrementalLoadOffsetValue(resultSetEndReached);
      context.commitOffset(tableRuntimeContext.getOffsetKey(), offsetValue);
    }

    //Make sure we close the result set only when there are no more rows in the result set
    if (shouldEvict.get()) {
      //Invalidate so as to fetch a new result set
      //We close the result set/statement in Removal Listener
      tableReadContextCache.invalidate(tableRuntimeContext);

      tableProvider.releaseOwnedTable(tableRuntimeContext, threadNumber);

      tableRuntimeContext = null;
    } else if (
        tableJdbcConfigBean.batchTableStrategy == BatchTableStrategy.SWITCH_TABLES
            && !tableRuntimeContext.isUsingNonIncrementalLoad()) {
      tableRuntimeContext = null;
    }

    final List<TableRuntimeContext> removedPartitions = tableProvider.getAndClearRemovedPartitions();
    if (removedPartitions != null && removedPartitions.size() > 0) {
      for (TableRuntimeContext partition : removedPartitions) {
        LOG.debug(
            "Removing offset entry for partition {} since it has been removed from the table provider",
            partition.getDescription()
        );
        context.commitOffset(partition.getOffsetKey(), null);
      }
    }
  }
After a batch is generate perform needed operations, generate and commit batch
Evict entries from {@link #tableReadContextCache}, reset {@link #tableRuntimeContext}