[flink] Support custom properties in Flink table factory

[LiebingYu]

Purpose

Linked issue: close #2425

Brief change log

Tests

API and Format

Documentation

[luoyuxia]

@LiebingYu Thanks for thr pr. LGTM @loserwang1024 Could you please help review again?

[loserwang1024]

@LiebingYu Should we add a specially prefix to custom properties rather than skip all the check? I think it's dangerous for later compability.

Currently org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils#validateTableSourceOptions only check the scan start up.

why not when catalog read from table, add a prefixes 'customer.' to customers. Then table factory check it and remove the prefixes? @leonardBang @wuchong , CC, WDYT

[leonardBang]

@LiebingYu Should we add a specially prefix to custom properties rather than skip all the check? I think it's dangerous for later compability.

+1 for adding a specific prefix for custom properties instead of removing all existing validation for near all options, it make the code being weak. Imaging kafka and MySQL CDC connector, we also support custom properties with prefix i.e.:
Kafka：https://nightlies.apache.org/flink/flink-docs-release-2.2/docs/connectors/table/kafka/#properties

[wuchong]

I think it’s a good idea to use a dedicated prefix for user-extensible properties to avoid potential key conflicts in the future.

Instead of customer.*, consider clearer and more conventional prefixes like tags.*, ext.*, or properties.*, or even begins with _ prefix such as _owner=xxx. These better convey that the keys represent custom, user-defined metadata. The term customer is ambiguous and doesn’t clearly signal extensibility. Moreover, from Fluss’s perspective, any property outside the table.* namespace is effectively a custom property.

Examples:

tags.owner = 'xxx'
ext.owner = 'xxx'
properties.owner = 'xxx'
_owner = 'xxx'

[wuchong]

Personally, I prefer tags.* because it clearly conveys that these are purely custom labels with no semantic impact on connector or storage behavior.

[LiebingYu]

I think it’s a good idea to use a dedicated prefix for user-extensible properties to avoid potential key conflicts in the future.

Instead of customer.*, consider clearer and more conventional prefixes like tags.*, ext.*, or properties.*, or even begins with _ prefix such as _owner=xxx. These better convey that the keys represent custom, user-defined metadata. The term customer is ambiguous and doesn’t clearly signal extensibility. Moreover, from Fluss’s perspective, any property outside the table.* namespace is effectively a custom property.

Examples:

tags.owner = 'xxx'
ext.owner = 'xxx'
properties.owner = 'xxx'
_owner = 'xxx'

My question is: do we really need to define a prefix for custom properties? As you said, “from Fluss’s perspective, any property outside the table.* namespace is effectively a custom property.” When the Flink connector processes properties, it can filter out those custom ones and simply skip validating them. From this perspective, custom properties don’t necessarily need a prefix, right? After all, the Fluss server imposes no restrictions on custom properties, but the Flink read/write path introduces an additional constraint (and if users create tables via the API, they can completely bypass this constraint).