[core] Support watermark partition markdone mode

[xiangyuf]

Purpose

Linked issue: close #4963

Use watermark to trigger partition markdone

Tests

E2E Test in WatermarkPartitionMarkDoneTest#testWaterMarkPartitionMarkDone

API and Format

Documentation

[LinMingQiang]

You should ignore partition.time-interval, otherwise watermark will not be able to properly determine when the partition should be mark done.

[xiangyuf]

You should ignore partition.time-interval, otherwise watermark will not be able to properly determine when the partition should be mark done.

@LinMingQiang Thx for advice. I don't think the partition.time-interval will affect the watermark partition mark done behavior here. Can you explain more here?

[LinMingQiang]

You should ignore partition.time-interval, otherwise watermark will not be able to properly determine when the partition should be mark done.

@LinMingQiang Thx for advice. I don't think the partition.time-interval will affect the watermark partition mark done behavior here. Can you explain more here?

e.g. partition.time-interval = 2 d ，partition value is '2025-03-10', watermark is '2025-03-11'. Should partition '2025-03-10' be marked as done? The current logic is not.

Related code :

Map.Entry<String, Long> entry = iter.next();
            String partition = entry.getKey();

            long lastUpdateTime = entry.getValue();
            long partitionStartTime =
                    extractDateTime(partition)
                            .atZone(ZoneId.systemDefault())
                            .toInstant()
                            .toEpochMilli();
            long partitionEndTime = partitionStartTime + timeInterval;
            lastUpdateTime = Math.max(lastUpdateTime, partitionEndTime);

            if (currentTimeMillis - lastUpdateTime > idleTime) {
                needDone.add(partition);
                iter.remove();
            }

[xiangyuf]

You should ignore partition.time-interval, otherwise watermark will not be able to properly determine when the partition should be mark done.

@LinMingQiang Thx for advice. I don't think the partition.time-interval will affect the watermark partition mark done behavior here. Can you explain more here?

e.g. partition.time-interval = 2 d ，partition value is '2025-03-10', watermark is '2025-03-11'. Should partition '2025-03-10' be marked as done? The current logic is not.

Related code :

Map.Entry<String, Long> entry = iter.next();
            String partition = entry.getKey();

            long lastUpdateTime = entry.getValue();
            long partitionStartTime =
                    extractDateTime(partition)
                            .atZone(ZoneId.systemDefault())
                            .toInstant()
                            .toEpochMilli();
            long partitionEndTime = partitionStartTime + timeInterval;
            lastUpdateTime = Math.max(lastUpdateTime, partitionEndTime);

            if (currentTimeMillis - lastUpdateTime > idleTime) {
                needDone.add(partition);
                iter.remove();
            }

In your case, I think partition '2025-03-10' should be mark done after watermark pass '2025-03-12'.

[LinMingQiang]

You should ignore partition.time-interval, otherwise watermark will not be able to properly determine when the partition should be mark done.

@LinMingQiang Thx for advice. I don't think the partition.time-interval will affect the watermark partition mark done behavior here. Can you explain more here?

e.g. partition.time-interval = 2 d ，partition value is '2025-03-10', watermark is '2025-03-11'. Should partition '2025-03-10' be marked as done? The current logic is not.
Related code :

Map.Entry<String, Long> entry = iter.next();
            String partition = entry.getKey();

            long lastUpdateTime = entry.getValue();
            long partitionStartTime =
                    extractDateTime(partition)
                            .atZone(ZoneId.systemDefault())
                            .toInstant()
                            .toEpochMilli();
            long partitionEndTime = partitionStartTime + timeInterval;
            lastUpdateTime = Math.max(lastUpdateTime, partitionEndTime);

            if (currentTimeMillis - lastUpdateTime > idleTime) {
                needDone.add(partition);
                iter.remove();
            }

In your case, I think partition '2025-03-10' should be mark done after watermark pass '2025-03-12'.

👌.

[xiangyuf]

@JingsongLi Ready for review

[JingsongLi]

+1