code: public <T extends Watermark> T getActualHighWatermark(Class<T> watermarkClass, Gson gson) {
      JsonElement json = getActualHighWatermark();
      if (json == null) {
        json = this.workUnit.getLowWatermark();
        if (json == null) {
          return null;
        }
      }
      return gson.fromJson(json, watermarkClass);
    }
docs: Get the actual high {@link Watermark}. If the {@code WorkUnitState} does not contain the actual high watermark
  (which may be caused by task failures), the low watermark in the corresponding {@link WorkUnit} will be returned.
  
  @param watermarkClass the watermark class for this {@code WorkUnitState}.
  @param gson a {@link Gson} object used to deserialize the watermark.
  @return the actual high watermark in this {@code WorkUnitState}. null is returned if this {@code WorkUnitState}
  does not contain an actual high watermark, and the corresponding {@code WorkUnit} does not contain a low
  watermark.