[flink] Auto create new tag for flink checkpoint

[MonsterChenzhuo]

Purpose

close #1942

The generation of tags is driven by checkpoints. Although no new data is written to trigger the creation of a new snapshot, tags can still be generated and point to the current latest snapshot.

Tests

API and Format

Documentation

[JingsongLi]

remove this sentence.. there is no sink.savepoint.auto-tag here.

Document after You can set sink.savepoint.auto-tagtotrue to enable the feature of automatically creating tags for savepoint.

[MonsterChenzhuo]

done.

[JingsongLi]

Just simplify documentation. For example: This option will also force creating a Tag for the tag.automatic-creation=process-time mode at an appropriate time even when no data is written.

[MonsterChenzhuo]

done.

[JingsongLi]

Don't remove this empty line, this is just personal code style

[MonsterChenzhuo]

done.

[JingsongLi]

Why add this?

[MonsterChenzhuo]

done.

[JingsongLi]

We should not add new method for TagAutoCreation.
We should just invoke run.

We can create a method:

private void tryToTag(Snapshot snapshot, LocalDateTime time) {
        if (nextTag == null
                || isAfterOrEqual(time.minus(delay), periodHandler.nextTagTime(nextTag))) {
            LocalDateTime thisTag = periodHandler.normalizeToTagTime(time);
            String tagName = periodHandler.timeToTag(thisTag);
            tagManager.createTag(snapshot, tagName);
            nextTag = periodHandler.nextTagTime(thisTag);

            if (numRetainedMax != null) {
                SortedMap<Snapshot, String> tags = tagManager.tags();
                if (tags.size() > numRetainedMax) {
                    int toDelete = tags.size() - numRetainedMax;
                    int i = 0;
                    for (String tag : tags.values()) {
                        tagManager.deleteTag(tag, tagDeletion, snapshotManager);
                        i++;
                        if (i == toDelete) {
                            break;
                        }
                    }
                }
            }
        }
    }

And in run:

    public void run() {
        while (true) {
            if (snapshotManager.snapshotExists(nextSnapshot)) {
                tryToTag(snapshotManager.snapshot(nextSnapshot));
                nextSnapshot++;
            } else {
                // avoid snapshot has been expired
                Long earliest = snapshotManager.earliestSnapshotId();
                if (earliest != null && earliest > nextSnapshot) {
                    nextSnapshot = earliest;
                } else {
                    break;
                }
            }
        }

        if (timeExtractor instanceof ProcessTimeExtractor) {
            tryToTag(snapshotManager.snapshot(nextSnapshot - 1), LocalDateTime.now());
        }
    }

Something like above codes.

[MonsterChenzhuo]

done.

[JingsongLi]

Maybe you can try to extract tagAutoCreation from CommitterOperator. It is better to maintain only one object for in the committer.

[MonsterChenzhuo]

done.

[MonsterChenzhuo]

@JingsongLi We conducted a 10-day online test, and the results met our expectations.

[JingsongLi]

I think we can use a generic way to implement this.
For example, if the extractor is ProcessTimeExtractor in TagAutoCreation, we can launch a async thread to check tag creation.

[JingsongLi]

When I developed #2134
I found that snapshots are our fundamental solution, and another way to solve this problem is to force the creation of snapshots for tables (in this configuration case, or add a new configuration option), which is a relatively simple thing.

[JingsongLi]

#2188 should fixes this.