Sink Properties per Table

[arielmoraes]

What has changed

The current connector version doesn't support setting sink.properties.* per table. That defeats the purpose of having multiple topics in the same connector, which is very good in a multi-tenant environment.

This PR aims to change that by creating a new config prefix and connector (StarRocksSinkConnectorV2), avoiding breaking changes.

How it works

The solution uses a different approach compared to the original Sink as the properties are tied to the Load Manager.

Sink Properties

The properties are captured by using the following regex pattern: sink\.table\.(.+)\.properties\.(.+). That will capture properties like sink.table.my_table.properties.columns.

The properties are constructed by merging the default properties coming from the sink.properties.*, so if the default properties set has a columns property it would be replaced by the example above.

Load Manager

A Load Manager will be created for each table found when mapping the properties.

The put method will find the correct Load Manager by using the topic to table mapping, fallbacking to the default loader if not found.

The preCommit method will flush all Load Managers and abort on the first chance of exception.

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}