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);
        }
      }
    }
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.
Generate a batch ( looping through as many tables as needed ) until a batch can be generated and then commit offset .