code: 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;
    }
docs: Adjust a watermark based on watermark type
  
  @param baseWatermark the original watermark
  @param watermarkType Watermark Type
  @return the adjusted watermark value