@ Deprecated public HashMap < Long , Long > getPartitions ( long previousWatermark ) { HashMap < Long , Long > defaultPartition = Maps . newHashMap ( ) ; if ( ! isWatermarkExists ( ) ) { defaultPartition . put ( ConfigurationKeys . DEFAULT_WATERMARK_VALUE , ConfigurationKeys . DEFAULT_WATERMARK_VALUE ) ; LOG . info ( "Watermark column or type not found - Default partition with low watermark and high watermark as " + ConfigurationKeys . DEFAULT_WATERMARK_VALUE ) ; return defaultPartition ; } ExtractType extractType = ExtractType . valueOf ( this . state . getProp ( ConfigurationKeys . SOURCE_QUERYBASED_EXTRACT_TYPE ) . toUpperCase ( ) ) ; WatermarkType watermarkType = WatermarkType . valueOf ( this . state . getProp ( ConfigurationKeys . SOURCE_QUERYBASED_WATERMARK_TYPE , ConfigurationKeys . DEFAULT_WATERMARK_TYPE ) . toUpperCase ( ) ) ; int interval = getUpdatedInterval ( this . state . getPropAsInt ( ConfigurationKeys . SOURCE_QUERYBASED_PARTITION_INTERVAL , 0 ) , extractType , watermarkType ) ; int sourceMaxAllowedPartitions = this . state . getPropAsInt ( ConfigurationKeys . SOURCE_MAX_NUMBER_OF_PARTITIONS , 0 ) ; int maxPartitions = ( sourceMaxAllowedPartitions != 0 ? sourceMaxAllowedPartitions : ConfigurationKeys . DEFAULT_MAX_NUMBER_OF_PARTITIONS ) ; WatermarkPredicate watermark = new WatermarkPredicate ( null , watermarkType ) ; int deltaForNextWatermark = watermark . getDeltaNumForNextWatermark ( ) ; LOG . info ( "is watermark override: " + this . isWatermarkOverride ( ) ) ; LOG . info ( "is full extract: " + this . isFullDump ( ) ) ; long lowWatermark = this . getLowWatermark ( extractType , watermarkType , previousWatermark , deltaForNextWatermark ) ; long highWatermark = this . getHighWatermark ( extractType , watermarkType ) ; if ( lowWatermark == ConfigurationKeys . DEFAULT_WATERMARK_VALUE || highWatermark == ConfigurationKeys . DEFAULT_WATERMARK_VALUE ) { LOG . info ( "Low watermark or high water mark is not found. Hence cannot generate partitions - Default partition with low watermark: " + lowWatermark + " and high watermark: " + highWatermark ) ; defaultPartition . put ( lowWatermark , highWatermark ) ; return defaultPartition ; } LOG . info ( "Generate partitions with low watermark: " + lowWatermark + "; high watermark: " + highWatermark + "; partition interval in hours: " + interval + "; Maximum number of allowed partitions: " + maxPartitions ) ; return watermark . getPartitions ( lowWatermark , highWatermark , interval , maxPartitions ) ; }
Get partitions with low and high water marks