jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/multithread/MultithreadedTableProvider.java
@VisibleForTesting
  void acquireTableAsNeeded(int threadNumber) throws InterruptedException {
    if (!getOwnedTablesQueue().isEmpty() && batchTableStrategy == BatchTableStrategy.SWITCH_TABLES) {
      final TableRuntimeContext lastOwnedPartition = getOwnedTablesQueue().pollLast();


      if (getTableContextMap().containsValue(lastOwnedPartition.getSourceTableContext())) {
        sharedAvailableTablesQueue.offer(lastOwnedPartition);
      }

      TableContext lastOwnedTable = lastOwnedPartition.getSourceTableContext();
      // need to cycle off all partitions from the same table to the end of the queue
      TableRuntimeContext first = sharedAvailableTablesQueue.peek();
      while (first != null && first.getSourceTableContext().equals(lastOwnedTable)
          && !first.equals(lastOwnedPartition)) {
        if (LOG.isDebugEnabled()) {
          LOG.debug(
              "Moving partition {} to end of shared queue to comply with BatchTableStrategy of {}",
              first.getDescription(),
              batchTableStrategy.getLabel()
          );
        }

        // poll() should never return null since it is actually returning 'first' as we want to move it from the head
        // of the queue, that's why we are calling offer, to basically remove it from the head but keep it in the queue
        TableRuntimeContext toMove = sharedAvailableTablesQueue.poll();
        sharedAvailableTablesQueue.offer(toMove);
        // Get the new head of the queue
        first = sharedAvailableTablesQueue.peek();
      }
    }

    if (getOwnedTablesQueue().isEmpty()) {
      TableRuntimeContext head = sharedAvailableTablesQueue.poll();
      if (head != null) {
        offerToOwnedTablesQueue(head, threadNumber);
      }
    }

    partitionFirstSharedQueueItemIfNeeded();
  }
@VisibleForTesting
  void acquireTableAsNeeded(int threadNumber) throws InterruptedException {
    if (!getOwnedTablesQueue().isEmpty() && batchTableStrategy == BatchTableStrategy.SWITCH_TABLES) {
      final TableRuntimeContext lastOwnedPartition = getOwnedTablesQueue().pollLast();


      if (getTableContextMap().containsValue(lastOwnedPartition.getSourceTableContext())) {
        sharedAvailableTablesQueue.offer(lastOwnedPartition);
      }

      TableContext lastOwnedTable = lastOwnedPartition.getSourceTableContext();
      // need to cycle off all partitions from the same table to the end of the queue
      TableRuntimeContext first = sharedAvailableTablesQueue.peek();
      while (first != null && first.getSourceTableContext().equals(lastOwnedTable)
          && !first.equals(lastOwnedPartition)) {
        if (LOG.isDebugEnabled()) {
          LOG.debug(
              "Moving partition {} to end of shared queue to comply with BatchTableStrategy of {}",
              first.getDescription(),
              batchTableStrategy.getLabel()
          );
        }

        // poll() should never return null since it is actually returning 'first' as we want to move it from the head
        // of the queue, that's why we are calling offer, to basically remove it from the head but keep it in the queue
        TableRuntimeContext toMove = sharedAvailableTablesQueue.poll();
        sharedAvailableTablesQueue.offer(toMove);
        // Get the new head of the queue
        first = sharedAvailableTablesQueue.peek();
      }
    }

    if (getOwnedTablesQueue().isEmpty()) {
      TableRuntimeContext head = sharedAvailableTablesQueue.poll();
      if (head != null) {
        offerToOwnedTablesQueue(head, threadNumber);
      }
    }

    partitionFirstSharedQueueItemIfNeeded();
  }
@ VisibleForTesting void acquireTableAsNeeded ( int threadNumber ) throws InterruptedException { if ( ! getOwnedTablesQueue ( ) . isEmpty ( ) && batchTableStrategy == BatchTableStrategy . SWITCH_TABLES ) { final TableRuntimeContext lastOwnedPartition = getOwnedTablesQueue ( ) . pollLast ( ) ; if ( getTableContextMap ( ) . containsValue ( lastOwnedPartition . getSourceTableContext ( ) ) ) { sharedAvailableTablesQueue . offer ( lastOwnedPartition ) ; } TableContext lastOwnedTable = lastOwnedPartition . getSourceTableContext ( ) ; // need to cycle off all partitions from the same table to the end of the queue TableRuntimeContext first = sharedAvailableTablesQueue . peek ( ) ; while ( first != null && first . getSourceTableContext ( ) . equals ( lastOwnedTable ) && ! first . equals ( lastOwnedPartition ) ) { if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Moving partition {} to end of shared queue to comply with BatchTableStrategy of {}" , first . getDescription ( ) , batchTableStrategy . getLabel ( ) ) ; } // poll() should never return null since it is actually returning 'first' as we want to move it from the head // of the queue, that's why we are calling offer, to basically remove it from the head but keep it in the queue TableRuntimeContext toMove = sharedAvailableTablesQueue . poll ( ) ; sharedAvailableTablesQueue . offer ( toMove ) ; // Get the new head of the queue first = sharedAvailableTablesQueue . peek ( ) ; } } if ( getOwnedTablesQueue ( ) . isEmpty ( ) ) { TableRuntimeContext head = sharedAvailableTablesQueue . poll ( ) ; if ( head != null ) { offerToOwnedTablesQueue ( head , threadNumber ) ; } } partitionFirstSharedQueueItemIfNeeded ( ) ; }
Basically acquires more tables for the current thread to work on.
The maximum a thread can hold is upper bounded to the
value the thread number was allocated in {@link #threadNumToMaxTableSlots}
If there are no tables currently owned make a blocking call to {@link #sharedAvailableTablesQueue}
else simply poll {@link #sharedAvailableTablesQueue} and it to the {@link #ownedTablesQueue}
Basically acquires more tables for the current thread to work on . The maximum a thread can hold is upper bounded to the value the thread number was allocated in {