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()
      );
    }
  }
Each {@link TableJdbcRunnable} worker thread can call this api to update
if there is data/no more data on the current table