private static long adjustWatermark ( String baseWatermark , WatermarkType watermarkType ) { long result = ConfigurationKeys . DEFAULT_WATERMARK_VALUE ; switch ( watermarkType ) { case SIMPLE : result = SimpleWatermark . adjustWatermark ( baseWatermark , 0 ) ; break ; case DATE : result = DateWatermark . adjustWatermark ( baseWatermark , 0 ) ; break ; case HOUR : result = HourWatermark . adjustWatermark ( baseWatermark , 0 ) ; break ; case TIMESTAMP : result = TimestampWatermark . adjustWatermark ( baseWatermark , 0 ) ; break ; } return result ; }
Adjust a watermark based on watermark type