[FLINK-39777] Support configurable HashFunction strategies in PrePartitionOperator

[haruki-830]

Summary

This commit adds configurable partitioning strategy support to Flink CDC pipelines, enabling users to switch to table-id hashing for small tables via YAML configuration, reducing unnecessary primary-key hashing overhead.

Key Changes

1. New HashFunctionStrategy Enum

• Introduced HashFunctionStrategy enum with two options: PRIMARY_KEY (hash by TableId + primary keys) and TABLE_ID (hash by TableId only).
• Designed with @PublicEvolving annotation, allowing future strategies like ROUND_ROBIN or COLUMNS.

2. New TableIdHashFunctionProvider

• Added TableIdHashFunctionProvider that computes hash based solely on TableId.
• Uses singleton pattern for HashFunction since it is stateless.
• Suitable for small tables, or tables with no/changing primary keys.

3. Pipeline Configuration Option

• Added pipeline option "partitioning.strategy" to allow switching strategies via YAML.
• When unset, behavior is identical to current, ensuring full backward compatibility.

4. Comprehensive Testing

• Added TableIdHashFunctionProviderTest with 7 test.
• Added PrePartitionOperatorTest case verifying TABLE_ID strategy.
• Added FlinkPipelineComposerTest configurations for PRIMARY_KEY and TABLE_ID strategy validation.

Configuration Example

Route same-table events to a single subtask

pipeline:
  name: my-cdc-job
  partitioning.strategy: TABLE_ID

Force load-balanced distribution by primary keys

pipeline:
  name: my-cdc-job
  partitioning.strategy: PRIMARY_KEY

JIRA Reference

https://issues.apache.org/jira/browse/FLINK-39777

[haruki-830]

@lvyanquan could you please help review this PR?

[lvyanquan]

Thanks @haruki-830 for the feature, left some comments.

[lvyanquan]

Please update the document to guide users on how to use it.

[lvyanquan]

Please also update https://github.com/apache/flink-cdc/blob/master/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full.yaml#L56 / https://github.com/apache/flink-cdc/blob/master/flink-cdc-cli/src/test/resources/definitions/pipeline-definition-full-with-repsym.yaml#L59 and related tests.

[lvyanquan]

Using TableId#hashCode directly instead of Objects.hash(objectsToHash.toArray()) would be simpler.