void checkForInvalidInitialOffsetValues(
      PushSource.Context context,
      List<Stage.ConfigIssue> issues,
      String qualifiedTableName,
      LinkedHashMap<String, Integer> offsetColumnToType,
      Map<String, String> offsetColumnToStartOffset
  ) throws StageException {
    List<String> invalidInitialOffsetFieldAndValue =  new ArrayList<>();
    offsetColumnToType.forEach((offsetColumn, offsetSqlType) -> {
      String initialOffsetValue = offsetColumnToStartOffset.get(offsetColumn);
      try {
        if (jdbcUtil.isSqlTypeOneOf(offsetSqlType, Types.DATE, Types.TIME, Types.TIMESTAMP)) {
          Long.valueOf(initialOffsetValue); //NOSONAR
        } else {
          //Use native field conversion strategy to conver string to specify type and get value
          Field.create(OffsetQueryUtil.SQL_TYPE_TO_FIELD_TYPE.get(offsetSqlType), initialOffsetValue).getValue();
        }
      } catch (IllegalArgumentException e) {
        LOG.error(
            Utils.format(
                "Invalid Initial Offset Value {} for column {} in table {}",
                initialOffsetValue,
                offsetColumn,
                qualifiedTableName
            ),
            e
        );
        invalidInitialOffsetFieldAndValue.add(offsetColumn + " - " + initialOffsetValue);
      }
    });
    if (!invalidInitialOffsetFieldAndValue.isEmpty()) {
      throw new StageException(
          JdbcErrors.JDBC_72,
          qualifiedTableName,
          COMMA_JOINER.join(invalidInitialOffsetFieldAndValue)
      );
    }
  }
void checkForInvalidInitialOffsetValues ( PushSource . Context context , List < Stage . ConfigIssue > issues , String qualifiedTableName , LinkedHashMap < String , Integer > offsetColumnToType , Map < String , String > offsetColumnToStartOffset ) throws StageException { List < String > invalidInitialOffsetFieldAndValue = new ArrayList < > ( ) ; offsetColumnToType . forEach ( ( offsetColumn , offsetSqlType ) -> { String initialOffsetValue = offsetColumnToStartOffset . get ( offsetColumn ) ; try { if ( jdbcUtil . isSqlTypeOneOf ( offsetSqlType , Types . DATE , Types . TIME , Types . TIMESTAMP ) ) { Long . valueOf ( initialOffsetValue ) ; } else { Field . create ( OffsetQueryUtil . SQL_TYPE_TO_FIELD_TYPE . get ( offsetSqlType ) , initialOffsetValue ) . getValue ( ) ; } } catch ( IllegalArgumentException e ) { LOG . error ( Utils . format ( "Invalid Initial Offset Value {} for column {} in table {}" , initialOffsetValue , offsetColumn , qualifiedTableName ) , e ) ; invalidInitialOffsetFieldAndValue . add ( offsetColumn + " - " + initialOffsetValue ) ; } } ) ; if ( ! invalidInitialOffsetFieldAndValue . isEmpty ( ) ) { throw new StageException ( JdbcErrors . JDBC_72 , qualifiedTableName , COMMA_JOINER . join ( invalidInitialOffsetFieldAndValue ) ) ; } }
@VisibleForTesting