[core] Add custom tag expiration strategy

[Akwangg]

Purpose

Currently, there are some problems in tag expiration mechanism:

1. Normally, the expiration in the commit phase is based only on the tag retention time, and those tags without retention time are not processed, which needs to be expired separately.
2. When setting tag.automatic-creation=batch, it will first expire once in the commit phase, and then call its own separate expiration logic, which deletes only based on tag number, which will force the deletion of tags that have not expired, it is not reasonable.

so, we add the tag expiration strategy, it support three strategies:

• retain-time: expire only based on tag retain time.
• retain-number: expire only base on tag retain number even if the tag retain time is not reached. This strategy need set conf tag.num-retained-max.
• hybrid: first expire based the tag retain time, then consider retain number if conf tag.num-retained-max is set.

If we does not set tag.num-retained-max, the hybrid strategy is equivalent to retain-time strategy.

Tests

API and Format

Documentation

[JingsongLi]

Hi @askwang , Why not set another value to be very large so that it won't take effect

[JingsongLi]

Regarding the cleaning of the number of tags:

TagAutoCreation will clean up numbers and check if they are automatically created tags. It is difficult to adjust for streaming tasks as there may be compatibility issues.

We won't make any adjustments for now. If there is any new information, please submit a new PR.