jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/multithread/util/OffsetQueryUtil.java
public static Map<String, String> validateStoredAndSpecifiedOffset(TableContext tableContext, String offset) throws StageException {
    Set<String> expectedColumns = Sets.newHashSet(tableContext.getOffsetColumns());
    final Map<String, String> actualOffsets = getColumnsToOffsetMapFromOffsetFormat(offset);

    // only perform the actual validation below if there ARE stored offsets
    if (actualOffsets.size() == 0) {
      return actualOffsets;
    }

    Set<String> actualColumns = actualOffsets.keySet();

    Set<String> expectedSetDifference = Sets.difference(expectedColumns, actualColumns);
    Set<String> actualSetDifference = Sets.difference(actualColumns, expectedColumns);

    if (expectedSetDifference.size() > 0 || actualSetDifference.size() >  0) {
      throw new StageException(
          JdbcErrors.JDBC_71,
          tableContext.getQualifiedName(),
          COMMA_SPACE_JOINER.join(actualColumns),
          COMMA_SPACE_JOINER.join(expectedColumns)
      );
    }
    return actualOffsets;
  }
public static Map<String, String> validateStoredAndSpecifiedOffset(TableContext tableContext, String offset) throws StageException {
    Set<String> expectedColumns = Sets.newHashSet(tableContext.getOffsetColumns());
    final Map<String, String> actualOffsets = getColumnsToOffsetMapFromOffsetFormat(offset);

    // only perform the actual validation below if there ARE stored offsets
    if (actualOffsets.size() == 0) {
      return actualOffsets;
    }

    Set<String> actualColumns = actualOffsets.keySet();

    Set<String> expectedSetDifference = Sets.difference(expectedColumns, actualColumns);
    Set<String> actualSetDifference = Sets.difference(actualColumns, expectedColumns);

    if (expectedSetDifference.size() > 0 || actualSetDifference.size() >  0) {
      throw new StageException(
          JdbcErrors.JDBC_71,
          tableContext.getQualifiedName(),
          COMMA_SPACE_JOINER.join(actualColumns),
          COMMA_SPACE_JOINER.join(expectedColumns)
      );
    }
    return actualOffsets;
  }
Validates whether offset names match in the stored offset with respect to table configuration
@param tableContext {@link TableContext} for table
@param offset Stored offset from the previous stage run
@return the actual offset map as deserialized from the offset param, if valid
@throws StageException if columns mismatch