jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/multithread/MultithreadedTableProvider.java
public void reportDataOrNoMoreData(
      TableRuntimeContext tableRuntimeContext,
      int recordCount,
      int batchSize,
      boolean resultSetEndReached,
      AtomicBoolean tableFinished,
      AtomicBoolean schemaFinished,
      List<String> schemaFinishedTables
  ) {
    final TableContext sourceContext = tableRuntimeContext.getSourceTableContext();

    // When we see a table with data, we mark isNoMoreDataEventGeneratedAlready to false
    // so we can generate event again if we don't see data from all tables.
    if(recordCount > 0) {
      isNoMoreDataEventGeneratedAlready = false;
      tablesWithNoMoreData.remove(tableRuntimeContext.getSourceTableContext());
      remainingSchemasToTableContexts.put(sourceContext.getSchema(), sourceContext);
      completedSchemasToTableContexts.remove(sourceContext.getSchema(), sourceContext);
    }

    // we need to account for the activeRuntimeContexts here
    // if there are still other active contexts in process, then this should do "nothing"
    // if there are not other contexts, we need to figure out what the highest offset completed by the last batch was

    final boolean noMoreData = recordCount == 0 || resultSetEndReached;

    if (noMoreData) {
      tableRuntimeContext.setMarkedNoMoreData(true);
    }

    if (recordCount > 0) {
      maxPartitionWithDataPerTable.put(sourceContext, tableRuntimeContext.getPartitionSequence());
    }

    boolean tableExhausted = removePartitionIfNeeded(tableRuntimeContext);

    if (noMoreData) {
      if (tableExhausted) {
        synchronized (this) {
          if (LOG.isDebugEnabled()) {
            LOG.debug(
                "Table {} exhausted",
                sourceContext.getQualifiedName()
            );
          }

          final boolean newlyFinished = tablesWithNoMoreData.add(sourceContext);

          if (newlyFinished && tableFinished != null) {
            tableFinished.set(true);
          }

          final boolean remainingSchemaChanged = remainingSchemasToTableContexts.remove(sourceContext.getSchema(), sourceContext);
          completedSchemasToTableContexts.put(sourceContext.getSchema(), sourceContext);

          if (remainingSchemaChanged && remainingSchemasToTableContexts.get(sourceContext.getSchema()).isEmpty() && schemaFinished != null) {
            schemaFinished.set(true);
            if (schemaFinishedTables != null) {
              completedSchemasToTableContexts.get(sourceContext.getSchema()).forEach(
                  t -> schemaFinishedTables.add(t.getTableName())
              );
            }
          }
        }
      }
    }

    if (LOG.isTraceEnabled()) {
      LOG.trace(
          "Just released table {}; Number of Tables With No More Data {}",
          tableRuntimeContext.getDescription(),
          tablesWithNoMoreData.size()
      );
    }
  }
public void reportDataOrNoMoreData(
      TableRuntimeContext tableRuntimeContext,
      int recordCount,
      int batchSize,
      boolean resultSetEndReached,
      AtomicBoolean tableFinished,
      AtomicBoolean schemaFinished,
      List<String> schemaFinishedTables
  ) {
    final TableContext sourceContext = tableRuntimeContext.getSourceTableContext();

    // When we see a table with data, we mark isNoMoreDataEventGeneratedAlready to false
    // so we can generate event again if we don't see data from all tables.
    if(recordCount > 0) {
      isNoMoreDataEventGeneratedAlready = false;
      tablesWithNoMoreData.remove(tableRuntimeContext.getSourceTableContext());
      remainingSchemasToTableContexts.put(sourceContext.getSchema(), sourceContext);
      completedSchemasToTableContexts.remove(sourceContext.getSchema(), sourceContext);
    }

    // we need to account for the activeRuntimeContexts here
    // if there are still other active contexts in process, then this should do "nothing"
    // if there are not other contexts, we need to figure out what the highest offset completed by the last batch was

    final boolean noMoreData = recordCount == 0 || resultSetEndReached;

    if (noMoreData) {
      tableRuntimeContext.setMarkedNoMoreData(true);
    }

    if (recordCount > 0) {
      maxPartitionWithDataPerTable.put(sourceContext, tableRuntimeContext.getPartitionSequence());
    }

    boolean tableExhausted = removePartitionIfNeeded(tableRuntimeContext);

    if (noMoreData) {
      if (tableExhausted) {
        synchronized (this) {
          if (LOG.isDebugEnabled()) {
            LOG.debug(
                "Table {} exhausted",
                sourceContext.getQualifiedName()
            );
          }

          final boolean newlyFinished = tablesWithNoMoreData.add(sourceContext);

          if (newlyFinished && tableFinished != null) {
            tableFinished.set(true);
          }

          final boolean remainingSchemaChanged = remainingSchemasToTableContexts.remove(sourceContext.getSchema(), sourceContext);
          completedSchemasToTableContexts.put(sourceContext.getSchema(), sourceContext);

          if (remainingSchemaChanged && remainingSchemasToTableContexts.get(sourceContext.getSchema()).isEmpty() && schemaFinished != null) {
            schemaFinished.set(true);
            if (schemaFinishedTables != null) {
              completedSchemasToTableContexts.get(sourceContext.getSchema()).forEach(
                  t -> schemaFinishedTables.add(t.getTableName())
              );
            }
          }
        }
      }
    }

    if (LOG.isTraceEnabled()) {
      LOG.trace(
          "Just released table {}; Number of Tables With No More Data {}",
          tableRuntimeContext.getDescription(),
          tablesWithNoMoreData.size()
      );
    }
  }
public void reportDataOrNoMoreData ( TableRuntimeContext tableRuntimeContext , int recordCount , int batchSize , boolean resultSetEndReached , AtomicBoolean tableFinished , AtomicBoolean schemaFinished , List < String > schemaFinishedTables ) { final TableContext sourceContext = tableRuntimeContext . getSourceTableContext ( ) ; // When we see a table with data, we mark isNoMoreDataEventGeneratedAlready to false // so we can generate event again if we don't see data from all tables. if ( recordCount > 0 ) { isNoMoreDataEventGeneratedAlready = false ; tablesWithNoMoreData . remove ( tableRuntimeContext . getSourceTableContext ( ) ) ; remainingSchemasToTableContexts . put ( sourceContext . getSchema ( ) , sourceContext ) ; completedSchemasToTableContexts . remove ( sourceContext . getSchema ( ) , sourceContext ) ; } // we need to account for the activeRuntimeContexts here // if there are still other active contexts in process, then this should do "nothing" // if there are not other contexts, we need to figure out what the highest offset completed by the last batch was final boolean noMoreData = recordCount == 0 || resultSetEndReached ; if ( noMoreData ) { tableRuntimeContext . setMarkedNoMoreData ( true ) ; } if ( recordCount > 0 ) { maxPartitionWithDataPerTable . put ( sourceContext , tableRuntimeContext . getPartitionSequence ( ) ) ; } boolean tableExhausted = removePartitionIfNeeded ( tableRuntimeContext ) ; if ( noMoreData ) { if ( tableExhausted ) { synchronized ( this ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Table {} exhausted" , sourceContext . getQualifiedName ( ) ) ; } final boolean newlyFinished = tablesWithNoMoreData . add ( sourceContext ) ; if ( newlyFinished && tableFinished != null ) { tableFinished . set ( true ) ; } final boolean remainingSchemaChanged = remainingSchemasToTableContexts . remove ( sourceContext . getSchema ( ) , sourceContext ) ; completedSchemasToTableContexts . put ( sourceContext . getSchema ( ) , sourceContext ) ; if ( remainingSchemaChanged && remainingSchemasToTableContexts . get ( sourceContext . getSchema ( ) ) . isEmpty ( ) && schemaFinished != null ) { schemaFinished . set ( true ) ; if ( schemaFinishedTables != null ) { completedSchemasToTableContexts . get ( sourceContext . getSchema ( ) ) . forEach ( t -> schemaFinishedTables . add ( t . getTableName ( ) ) ) ; } } } } } if ( LOG . isTraceEnabled ( ) ) { LOG . trace ( "Just released table {}; Number of Tables With No More Data {}" , tableRuntimeContext . getDescription ( ) , tablesWithNoMoreData . size ( ) ) ; } }
Each {@link TableJdbcRunnable} worker thread can call this api to update
if there is data/no more data on the current table
Each {