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);
      }
    }
  }
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}
After a batch is generate perform needed operations generate and commit batch Evict entries from {