private long getSnapshotLowWatermark ( WatermarkType watermarkType , long previousWatermark , int deltaForNextWatermark ) { LOG . debug ( "Getting snapshot low water mark" ) ; String timeZone = this . state . getProp ( ConfigurationKeys . SOURCE_TIMEZONE , ConfigurationKeys . DEFAULT_SOURCE_TIMEZONE ) ; if ( isPreviousWatermarkExists ( previousWatermark ) ) { if ( isSimpleWatermark ( watermarkType ) ) { return previousWatermark + deltaForNextWatermark - this . state . getPropAsInt ( ConfigurationKeys . SOURCE_QUERYBASED_LOW_WATERMARK_BACKUP_SECS , 0 ) ; } DateTime wm = Utils . toDateTime ( previousWatermark , WATERMARKTIMEFORMAT , timeZone ) . plusSeconds ( ( deltaForNextWatermark - this . state . getPropAsInt ( ConfigurationKeys . SOURCE_QUERYBASED_LOW_WATERMARK_BACKUP_SECS , 0 ) ) ) ; return Long . parseLong ( Utils . dateTimeToString ( wm , WATERMARKTIMEFORMAT , timeZone ) ) ; } // If previous watermark is not found, override with the start value // (irrespective of source.is.watermark.override flag) long startValue = Utils . getLongWithCurrentDate ( this . state . getProp ( ConfigurationKeys . SOURCE_QUERYBASED_START_VALUE ) , timeZone ) ; LOG . info ( "Overriding low water mark with the given start value: " + startValue ) ; return startValue ; }
Get low water mark