@Override
  public short calculatePercentCompletion(Watermark lowWatermark, Watermark highWatermark) {
    Preconditions.checkArgument(
        lowWatermark instanceof MultiLongWatermark && highWatermark instanceof MultiLongWatermark,
        String.format("Arguments of %s.%s must be of type %s", MultiLongWatermark.class.getSimpleName(),
            Thread.currentThread().getStackTrace()[1].getMethodName(), MultiLongWatermark.class.getSimpleName()));

    long pulled = ((MultiLongWatermark) lowWatermark).getGap(this);
    long all = ((MultiLongWatermark) lowWatermark).getGap((MultiLongWatermark) highWatermark);
    Preconditions.checkState(all > 0);
    long percent = Math.min(100, LongMath.divide(pulled * 100, all, RoundingMode.HALF_UP));
    return (short) percent;
  }
@ Override public short calculatePercentCompletion ( Watermark lowWatermark , Watermark highWatermark ) { Preconditions . checkArgument ( lowWatermark instanceof MultiLongWatermark && highWatermark instanceof MultiLongWatermark , String . format ( "Arguments of %s.%s must be of type %s" , MultiLongWatermark . class . getSimpleName ( ) , Thread . currentThread ( ) . getStackTrace ( ) [ 1 ] . getMethodName ( ) , MultiLongWatermark . class . getSimpleName ( ) ) ) ; long pulled = ( ( MultiLongWatermark ) lowWatermark ) . getGap ( this ) ; long all = ( ( MultiLongWatermark ) lowWatermark ) . getGap ( ( MultiLongWatermark ) highWatermark ) ; Preconditions . checkState ( all > 0 ) ; long percent = Math . min ( 100 , LongMath . divide ( pulled * 100 , all , RoundingMode . HALF_UP ) ) ; return ( short ) percent ; }
Given a low watermark (starting point) and a high watermark (target), returns the percentage
of events pulled.

@return a percentage value between 0 and 100.