public Partition getGlobalPartition(long previousWatermark) {
    ExtractType extractType =
        ExtractType.valueOf(state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_EXTRACT_TYPE).toUpperCase());
    WatermarkType watermarkType = WatermarkType.valueOf(
        state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_WATERMARK_TYPE, ConfigurationKeys.DEFAULT_WATERMARK_TYPE)
            .toUpperCase());

    WatermarkPredicate watermark = new WatermarkPredicate(null, watermarkType);
    int deltaForNextWatermark = watermark.getDeltaNumForNextWatermark();

    long lowWatermark = getLowWatermark(extractType, watermarkType, previousWatermark, deltaForNextWatermark);
    long highWatermark = getHighWatermark(extractType, watermarkType);
    return new Partition(lowWatermark, highWatermark, true, hasUserSpecifiedHighWatermark);
  }
public Partition getGlobalPartition ( long previousWatermark ) { ExtractType extractType = ExtractType . valueOf ( state . getProp ( ConfigurationKeys . SOURCE_QUERYBASED_EXTRACT_TYPE ) . toUpperCase ( ) ) ; WatermarkType watermarkType = WatermarkType . valueOf ( state . getProp ( ConfigurationKeys . SOURCE_QUERYBASED_WATERMARK_TYPE , ConfigurationKeys . DEFAULT_WATERMARK_TYPE ) . toUpperCase ( ) ) ; WatermarkPredicate watermark = new WatermarkPredicate ( null , watermarkType ) ; int deltaForNextWatermark = watermark . getDeltaNumForNextWatermark ( ) ; long lowWatermark = getLowWatermark ( extractType , watermarkType , previousWatermark , deltaForNextWatermark ) ; long highWatermark = getHighWatermark ( extractType , watermarkType ) ; return new Partition ( lowWatermark , highWatermark , true , hasUserSpecifiedHighWatermark ) ; }
Get the global partition of the whole data set, which has the global low and high watermarks

@param previousWatermark previous watermark for computing the low watermark of current run
@return a Partition instance