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);
    }
  }
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)