jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/multithread/JdbcBaseRunnable.java
private void generateBatchAndCommitOffset(BatchContext batchContext) {
    int recordCount = 0;
    int eventCount = 0;
    try {
        while (tableRuntimeContext == null) {
          tableRuntimeContext = tableProvider.nextTable(threadNumber);
          if (tableRuntimeContext == null) {
            // small sleep before trying to acquire a table again, to potentially allow a new partition to be
            // returned to shared queue or created
            final boolean uninterrupted = ThreadUtil.sleep(ACQUIRE_TABLE_SLEEP_INTERVAL);
            if (!uninterrupted) {
              LOG.trace("Interrupted which trying to acquire table");
            }
            if (!uninterrupted || tableRuntimeContext == null) {
              return;
            }
          }
        }
        updateGauge(JdbcBaseRunnable.Status.QUERYING_TABLE);
        tableReadContext = getOrLoadTableReadContext();
        ResultSet rs = tableReadContext.getResultSet();
        boolean resultSetEndReached = false;
        try {
          updateGauge(JdbcBaseRunnable.Status.GENERATING_BATCH);
          while (recordCount < batchSize) {
            if (rs.isClosed() || !rs.next()) {
              if (rs.isClosed()) {
                LOG.trace("ResultSet is closed");
              }
              resultSetEndReached = true;
              break;
            }
            createAndAddRecord(rs, tableRuntimeContext, batchContext);
            recordCount++;
          }

          LOG.trace("{} records generated", recordCount);

          if (commonSourceConfigBean.enableSchemaChanges) {
            generateSchemaChanges(batchContext);
          }

          tableRuntimeContext.setResultSetProduced(true);

          //Reset numSqlErrors if we are able to read result set and add records to the batch context.
          numSQLErrors = 0;
          firstSqlException = null;

          //If exception happened we do not report anything about no more data event
          //We report noMoreData if either evictTableReadContext is true (result set no more rows) / record count is 0.

          final AtomicBoolean tableFinished = new AtomicBoolean(false);
          final AtomicBoolean schemaFinished = new AtomicBoolean(false);
          final List<String> schemaFinishedTables = new LinkedList<>();
          tableProvider.reportDataOrNoMoreData(
              tableRuntimeContext,
              recordCount,
              batchSize,
              resultSetEndReached,
              tableFinished,
              schemaFinished,
              schemaFinishedTables
          );

          if (tableFinished.get()) {
            TableJdbcEvents.createTableFinishedEvent(context, batchContext, tableRuntimeContext);
            eventCount++;
          }
          if (schemaFinished.get()) {
            TableJdbcEvents.createSchemaFinishedEvent(context, batchContext, tableRuntimeContext, schemaFinishedTables);
            eventCount++;
          }
        } finally {
          if (resultSetEndReached) {
            tableReadContext.closeResultSet();
          }
          handlePostBatchAsNeeded(resultSetEndReached, recordCount, eventCount, batchContext);
        }
      } catch (SQLException | ExecutionException | StageException | InterruptedException e) {
        LOG.error("Error happened", e);

        //invalidate if the connection is closed
        tableReadContextCache.invalidateAll();
        connectionManager.closeConnection();
        //If we have executed post batch that had errored out
        if (tableRuntimeContext != null) {
          final TableContext table = tableRuntimeContext.getSourceTableContext();
          //if the currently acquired tableContext is no longer a valid candidate, try re-fetch the table from table provider
          if (commonSourceConfigBean.allowLateTable && !tableProvider.getActiveRuntimeContexts().containsKey(table)) {
            tableRuntimeContext = null;
          }
        }

        Throwable th = (e instanceof ExecutionException)? e.getCause() : e;
        if (th instanceof SQLException) {
          handleSqlException((SQLException)th);
        } else if (e instanceof InterruptedException) {
          LOG.error("Thread {} interrupted", gaugeMap.get(THREAD_NAME));
        } else {
          handleStageError(JdbcErrors.JDBC_67, e);
        }
      }
    }
private void generateBatchAndCommitOffset(BatchContext batchContext) {
    int recordCount = 0;
    int eventCount = 0;
    try {
        while (tableRuntimeContext == null) {
          tableRuntimeContext = tableProvider.nextTable(threadNumber);
          if (tableRuntimeContext == null) {
            // small sleep before trying to acquire a table again, to potentially allow a new partition to be
            // returned to shared queue or created
            final boolean uninterrupted = ThreadUtil.sleep(ACQUIRE_TABLE_SLEEP_INTERVAL);
            if (!uninterrupted) {
              LOG.trace("Interrupted which trying to acquire table");
            }
            if (!uninterrupted || tableRuntimeContext == null) {
              return;
            }
          }
        }
        updateGauge(JdbcBaseRunnable.Status.QUERYING_TABLE);
        tableReadContext = getOrLoadTableReadContext();
        ResultSet rs = tableReadContext.getResultSet();
        boolean resultSetEndReached = false;
        try {
          updateGauge(JdbcBaseRunnable.Status.GENERATING_BATCH);
          while (recordCount < batchSize) {
            if (rs.isClosed() || !rs.next()) {
              if (rs.isClosed()) {
                LOG.trace("ResultSet is closed");
              }
              resultSetEndReached = true;
              break;
            }
            createAndAddRecord(rs, tableRuntimeContext, batchContext);
            recordCount++;
          }

          LOG.trace("{} records generated", recordCount);

          if (commonSourceConfigBean.enableSchemaChanges) {
            generateSchemaChanges(batchContext);
          }

          tableRuntimeContext.setResultSetProduced(true);

          //Reset numSqlErrors if we are able to read result set and add records to the batch context.
          numSQLErrors = 0;
          firstSqlException = null;

          //If exception happened we do not report anything about no more data event
          //We report noMoreData if either evictTableReadContext is true (result set no more rows) / record count is 0.

          final AtomicBoolean tableFinished = new AtomicBoolean(false);
          final AtomicBoolean schemaFinished = new AtomicBoolean(false);
          final List<String> schemaFinishedTables = new LinkedList<>();
          tableProvider.reportDataOrNoMoreData(
              tableRuntimeContext,
              recordCount,
              batchSize,
              resultSetEndReached,
              tableFinished,
              schemaFinished,
              schemaFinishedTables
          );

          if (tableFinished.get()) {
            TableJdbcEvents.createTableFinishedEvent(context, batchContext, tableRuntimeContext);
            eventCount++;
          }
          if (schemaFinished.get()) {
            TableJdbcEvents.createSchemaFinishedEvent(context, batchContext, tableRuntimeContext, schemaFinishedTables);
            eventCount++;
          }
        } finally {
          if (resultSetEndReached) {
            tableReadContext.closeResultSet();
          }
          handlePostBatchAsNeeded(resultSetEndReached, recordCount, eventCount, batchContext);
        }
      } catch (SQLException | ExecutionException | StageException | InterruptedException e) {
        LOG.error("Error happened", e);

        //invalidate if the connection is closed
        tableReadContextCache.invalidateAll();
        connectionManager.closeConnection();
        //If we have executed post batch that had errored out
        if (tableRuntimeContext != null) {
          final TableContext table = tableRuntimeContext.getSourceTableContext();
          //if the currently acquired tableContext is no longer a valid candidate, try re-fetch the table from table provider
          if (commonSourceConfigBean.allowLateTable && !tableProvider.getActiveRuntimeContexts().containsKey(table)) {
            tableRuntimeContext = null;
          }
        }

        Throwable th = (e instanceof ExecutionException)? e.getCause() : e;
        if (th instanceof SQLException) {
          handleSqlException((SQLException)th);
        } else if (e instanceof InterruptedException) {
          LOG.error("Thread {} interrupted", gaugeMap.get(THREAD_NAME));
        } else {
          handleStageError(JdbcErrors.JDBC_67, e);
        }
      }
    }
Generate a batch (looping through as many tables as needed) until a batch can be generated
and then commit offset.