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;
  }
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
Validates whether offset names match in the stored offset with respect to table configuration