jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/multithread/MultithreadedTableProvider.java
private void handlePartitioningTurnedOffOrOn(
      SortedSetMultimap<TableContext, TableRuntimeContext> reconstructedPartitions
  ) {

    for (TableContext tableContext : reconstructedPartitions.keySet()) {
      final SortedSet<TableRuntimeContext> partitions = reconstructedPartitions.get(tableContext);
      final TableRuntimeContext lastPartition = partitions.last();
      final TableContext sourceTableContext = lastPartition.getSourceTableContext();
      Utils.checkState(
          sourceTableContext.equals(tableContext),
          String.format(
              "Source table context for %s should match TableContext map key of %s",
              lastPartition.getDescription(),
              tableContext.getQualifiedName()
          )
      );

      final boolean partitioningTurnedOff = lastPartition.isPartitioned()
          && sourceTableContext.getPartitioningMode() == PartitioningMode.DISABLED;
      final boolean partitioningTurnedOn = !lastPartition.isPartitioned()
          && sourceTableContext.isPartitionable()
          && sourceTableContext.getPartitioningMode() != PartitioningMode.DISABLED;

      if (!partitioningTurnedOff && !partitioningTurnedOn) {
        continue;
      }

      final Map<String, String> nextStartingOffsets = new HashMap<>();
      final Map<String, String> nextMaxOffsets = new HashMap<>();

      final int newPartitionSequence = lastPartition.getPartitionSequence() > 0 ? lastPartition.getPartitionSequence() + 1 : 1;
      if (partitioningTurnedOff) {
        LOG.info(
            "Table {} has switched from partitioned to non-partitioned; partition sequence {} will be the last (with" +
                " no max offsets)",
            sourceTableContext.getQualifiedName(),
            newPartitionSequence
        );

        lastPartition.getStartingPartitionOffsets().forEach(
            (col, off) -> {
              String basedOnStartOffset = lastPartition.generateNextPartitionOffset(col, off);
              nextStartingOffsets.put(col, basedOnStartOffset);
            }
        );

      } else if (partitioningTurnedOn) {

        lastPartition.getStartingPartitionOffsets().forEach(
            (col, off) -> {
              String basedOnStoredOffset = lastPartition.getInitialStoredOffsets().get(col);
              nextStartingOffsets.put(col, basedOnStoredOffset);
            }
        );

        nextStartingOffsets.forEach(
            (col, off) -> nextMaxOffsets.put(col, lastPartition.generateNextPartitionOffset(col, off))
        );

        if (!reconstructedPartitions.remove(sourceTableContext, lastPartition)) {
          throw new IllegalStateException(String.format(
              "Failed to remove partition %s for table %s in switching partitioning from off to on",
              lastPartition.getDescription(),
              sourceTableContext.getQualifiedName()
          ));
        }

        LOG.info(
            "Table {} has switched from non-partitioned to partitioned; using last stored offsets as the starting" +
                " offsets for the new partition {}",
            sourceTableContext.getQualifiedName(),
            newPartitionSequence
        );
      }

      final TableRuntimeContext nextPartition = new TableRuntimeContext(
          sourceTableContext,
          lastPartition.isUsingNonIncrementalLoad(),
          (lastPartition.isPartitioned() && !partitioningTurnedOff) || partitioningTurnedOn,
          newPartitionSequence,
          nextStartingOffsets,
          nextMaxOffsets
      );

      reconstructedPartitions.put(sourceTableContext, nextPartition);
    }
  }
private void handlePartitioningTurnedOffOrOn(
      SortedSetMultimap<TableContext, TableRuntimeContext> reconstructedPartitions
  ) {

    for (TableContext tableContext : reconstructedPartitions.keySet()) {
      final SortedSet<TableRuntimeContext> partitions = reconstructedPartitions.get(tableContext);
      final TableRuntimeContext lastPartition = partitions.last();
      final TableContext sourceTableContext = lastPartition.getSourceTableContext();
      Utils.checkState(
          sourceTableContext.equals(tableContext),
          String.format(
              "Source table context for %s should match TableContext map key of %s",
              lastPartition.getDescription(),
              tableContext.getQualifiedName()
          )
      );

      final boolean partitioningTurnedOff = lastPartition.isPartitioned()
          && sourceTableContext.getPartitioningMode() == PartitioningMode.DISABLED;
      final boolean partitioningTurnedOn = !lastPartition.isPartitioned()
          && sourceTableContext.isPartitionable()
          && sourceTableContext.getPartitioningMode() != PartitioningMode.DISABLED;

      if (!partitioningTurnedOff && !partitioningTurnedOn) {
        continue;
      }

      final Map<String, String> nextStartingOffsets = new HashMap<>();
      final Map<String, String> nextMaxOffsets = new HashMap<>();

      final int newPartitionSequence = lastPartition.getPartitionSequence() > 0 ? lastPartition.getPartitionSequence() + 1 : 1;
      if (partitioningTurnedOff) {
        LOG.info(
            "Table {} has switched from partitioned to non-partitioned; partition sequence {} will be the last (with" +
                " no max offsets)",
            sourceTableContext.getQualifiedName(),
            newPartitionSequence
        );

        lastPartition.getStartingPartitionOffsets().forEach(
            (col, off) -> {
              String basedOnStartOffset = lastPartition.generateNextPartitionOffset(col, off);
              nextStartingOffsets.put(col, basedOnStartOffset);
            }
        );

      } else if (partitioningTurnedOn) {

        lastPartition.getStartingPartitionOffsets().forEach(
            (col, off) -> {
              String basedOnStoredOffset = lastPartition.getInitialStoredOffsets().get(col);
              nextStartingOffsets.put(col, basedOnStoredOffset);
            }
        );

        nextStartingOffsets.forEach(
            (col, off) -> nextMaxOffsets.put(col, lastPartition.generateNextPartitionOffset(col, off))
        );

        if (!reconstructedPartitions.remove(sourceTableContext, lastPartition)) {
          throw new IllegalStateException(String.format(
              "Failed to remove partition %s for table %s in switching partitioning from off to on",
              lastPartition.getDescription(),
              sourceTableContext.getQualifiedName()
          ));
        }

        LOG.info(
            "Table {} has switched from non-partitioned to partitioned; using last stored offsets as the starting" +
                " offsets for the new partition {}",
            sourceTableContext.getQualifiedName(),
            newPartitionSequence
        );
      }

      final TableRuntimeContext nextPartition = new TableRuntimeContext(
          sourceTableContext,
          lastPartition.isUsingNonIncrementalLoad(),
          (lastPartition.isPartitioned() && !partitioningTurnedOff) || partitioningTurnedOn,
          newPartitionSequence,
          nextStartingOffsets,
          nextMaxOffsets
      );

      reconstructedPartitions.put(sourceTableContext, nextPartition);
    }
  }
private void handlePartitioningTurnedOffOrOn ( SortedSetMultimap < TableContext , TableRuntimeContext > reconstructedPartitions ) { for ( TableContext tableContext : reconstructedPartitions . keySet ( ) ) { final SortedSet < TableRuntimeContext > partitions = reconstructedPartitions . get ( tableContext ) ; final TableRuntimeContext lastPartition = partitions . last ( ) ; final TableContext sourceTableContext = lastPartition . getSourceTableContext ( ) ; Utils . checkState ( sourceTableContext . equals ( tableContext ) , String . format ( "Source table context for %s should match TableContext map key of %s" , lastPartition . getDescription ( ) , tableContext . getQualifiedName ( ) ) ) ; final boolean partitioningTurnedOff = lastPartition . isPartitioned ( ) && sourceTableContext . getPartitioningMode ( ) == PartitioningMode . DISABLED ; final boolean partitioningTurnedOn = ! lastPartition . isPartitioned ( ) && sourceTableContext . isPartitionable ( ) && sourceTableContext . getPartitioningMode ( ) != PartitioningMode . DISABLED ; if ( ! partitioningTurnedOff && ! partitioningTurnedOn ) { continue ; } final Map < String , String > nextStartingOffsets = new HashMap <> ( ) ; final Map < String , String > nextMaxOffsets = new HashMap <> ( ) ; final int newPartitionSequence = lastPartition . getPartitionSequence ( ) > 0 ? lastPartition . getPartitionSequence ( ) + 1 : 1 ; if ( partitioningTurnedOff ) { LOG . info ( "Table {} has switched from partitioned to non-partitioned; partition sequence {} will be the last (with" + " no max offsets)" , sourceTableContext . getQualifiedName ( ) , newPartitionSequence ) ; lastPartition . getStartingPartitionOffsets ( ) . forEach ( ( col , off ) -> { String basedOnStartOffset = lastPartition . generateNextPartitionOffset ( col , off ) ; nextStartingOffsets . put ( col , basedOnStartOffset ) ; } ) ; } else if ( partitioningTurnedOn ) { lastPartition . getStartingPartitionOffsets ( ) . forEach ( ( col , off ) -> { String basedOnStoredOffset = lastPartition . getInitialStoredOffsets ( ) . get ( col ) ; nextStartingOffsets . put ( col , basedOnStoredOffset ) ; } ) ; nextStartingOffsets . forEach ( ( col , off ) -> nextMaxOffsets . put ( col , lastPartition . generateNextPartitionOffset ( col , off ) ) ) ; if ( ! reconstructedPartitions . remove ( sourceTableContext , lastPartition ) ) { throw new IllegalStateException ( String . format ( "Failed to remove partition %s for table %s in switching partitioning from off to on" , lastPartition . getDescription ( ) , sourceTableContext . getQualifiedName ( ) ) ) ; } LOG . info ( "Table {} has switched from non-partitioned to partitioned; using last stored offsets as the starting" + " offsets for the new partition {}" , sourceTableContext . getQualifiedName ( ) , newPartitionSequence ) ; } final TableRuntimeContext nextPartition = new TableRuntimeContext ( sourceTableContext , lastPartition . isUsingNonIncrementalLoad ( ) , ( lastPartition . isPartitioned ( ) && ! partitioningTurnedOff ) || partitioningTurnedOn , newPartitionSequence , nextStartingOffsets , nextMaxOffsets ) ; reconstructedPartitions . put ( sourceTableContext , nextPartition ) ; } }
Checks whether any tables have had partitioning turned off or not, and updates the partition map appropriately

@param reconstructedPartitions the reconstructed partitions (may be modified)
Checks whether any tables have had partitioning turned off or not and updates the partition map appropriately