public List < Partition > getPartitionList ( long previousWatermark ) { if ( state . getPropAsBoolean ( HAS_USER_SPECIFIED_PARTITIONS ) ) { return createUserSpecifiedPartitions ( ) ; } List < Partition > partitions = new ArrayList <> ( ) ; /* * Use the deprecated getPartitions(long) as a helper function, avoid duplicating logic. When it can be removed, its * logic will be put here. */ HashMap < Long , Long > partitionMap = getPartitions ( previousWatermark ) ; if ( partitionMap . size ( ) == 0 ) { return partitions ; } if ( partitionMap . size ( ) == 1 ) { Map . Entry < Long , Long > entry = partitionMap . entrySet ( ) . iterator ( ) . next ( ) ; Long lwm = entry . getKey ( ) ; Long hwm = entry . getValue ( ) ; if ( lwm == hwm ) { if ( lwm != - 1 ) { // we always allow [-1, -1] interval due to some test cases relies on this logic. boolean allowEqualBoundary = state . getPropAsBoolean ( ALLOW_EQUAL_WATERMARK_BOUNDARY , false ) ; LOG . info ( "Single partition with LWM = HWM and allowEqualBoundary=" + allowEqualBoundary ) ; if ( ! allowEqualBoundary ) { return partitions ; } } } } /* * Can't use highWatermark directly, as the partitionMap may have different precision. For example, highWatermark * may be specified to seconds, but partitionMap could be specified to hour or date. */ Long highestWatermark = Collections . max ( partitionMap . values ( ) ) ; for ( Map . Entry < Long , Long > entry : partitionMap . entrySet ( ) ) { Long partitionHighWatermark = entry . getValue ( ) ; // Apply hasUserSpecifiedHighWatermark to the last partition, which has highestWatermark if ( partitionHighWatermark . equals ( highestWatermark ) ) { partitions . add ( new Partition ( entry . getKey ( ) , partitionHighWatermark , true , hasUserSpecifiedHighWatermark ) ) ; } else { // The partitionHighWatermark was computed on the fly not what user specifies partitions . add ( new Partition ( entry . getKey ( ) , partitionHighWatermark , false ) ) ; } } return partitions ; }
Get an unordered list of partition with lowWatermark highWatermark and hasUserSpecifiedHighWatermark .