[FLINK-40051][pipeline-connector/hudi] Fix schema operator UID with operator UID prefix

[qiuyanjun888]

Summary

• Fix Hudi pipeline sink schema coordinator lookup when pipeline.operator.uid.prefix is configured.
• Add a regression test covering prefix-derived schema operator UID propagation into the Hudi sink.

Root Cause

FlinkPipelineComposer derives the schema operator UID from pipeline.operator.uid.prefix (for example, <prefix>-schema-operator) when the prefix option is set. However, HudiDataSinkFactory still passed pipeline.schema.operator.uid directly to HudiDataSink, so the Hudi sink used the deprecated default schema operator UID instead of the actual prefix-derived schema operator UID. This mismatch can make Hudi coordinator requests fail with No coordinator registered for operator.

Fix

• Make HudiDataSinkFactory derive the schema operator UID from pipeline.operator.uid.prefix when that option is explicitly configured.
• Preserve the existing pipeline.schema.operator.uid behavior when the prefix option is not set.
• Add HudiDataSinkFactoryTest#testSchemaOperatorUidFollowsOperatorUidPrefix.

Validation

• mvn -B -ntp -pl flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi -am -Dtest=HudiDataSinkFactoryTest#testSchemaOperatorUidFollowsOperatorUidPrefix -DfailIfNoTests=false test
• mvn -B -ntp -pl flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi -am -DfailIfNoTests=false test
• git diff --check

Jira

• https://issues.apache.org/jira/browse/FLINK-40051

Reviewers

Could @cshuo or @voonhous please review when convenient?

AI assistance

This PR was prepared with AI assistance. The focused regression test and Hudi module tests listed above were run locally.

[cshuo]

Could we avoid hard-coding the <prefix>-schema-operator derivation in the Hudi connector?

FlinkPipelineComposer already has the canonical UID generation logic, while Paimon/Iceberg/MaxCompute currently also read PIPELINE_SCHEMA_OPERATOR_UID directly before creating their SchemaEvolutionClient operator IDs. It may be better to move this final schema operator UID derivation into a shared helper in common and reuse it from both the composer and schema-aware sink factories, so the UID format does not drift across connectors.