[FLINK-36964] Fix potential exception when SchemaChange in parallel w…

[lvyanquan]

Fix potential exception when SchemaChange in parallel with Paimon Sink.

This close FLINK-36964 and FLINK-35888.

[yuxiqian]

Hi @lvyanquan, I have some concern about how BucketAssignOperator works with schema evolution stuff.

1. FlushEvent now works as a "pipeline drain indicator", which means after all data sink writer acknowledges them, there should't be any unhandled / uncommitted events flowing the whole pipeline, so SchemaRegistry could evolve downstream DB safely.

However, the bucket assigning strategy might break that assumption, where data sink writers might receive data change events with stale schema, even after external schema evolution processes have finished.

2. Potential schema operator hanging risks with "distributed" topology.

Basically same as described in #3680. In short, if a broadcast / custom partitioning topology is applied, then blocking one upstream partition will implicitly block all downstream partitions from handling events.

3. Why we need another Bucket assigner operator?

AFAIK all data change events have been hashed & distributed in PartitionOperator. Since adding a BucketAssignOperator does not change the parallelism, is there any reason we can't calculate the correct bucket partition ID in advance, instead of creating another partitioning topology?

[yuxiqian]

A little off topic, but we're really lacking E2e tests to real-life data sinks with complicated writing topologies. #3491 (with much higher testing pressure) might be necessary to expose more issues like this in the future.

[lvyanquan]

Hi @yuxiqian, could you help to review this?

[yuxiqian]

Thanks for @lvyanquan's great work, just left some comments.

[yuxiqian]

It's nice to have corresponding E2e cases to verify changes in this PR. Could @leonardBang @ruanhang1993 please trigger the CI workflow?

[lvyanquan]

Last CI passed and rebase master to fix the conflict.

[leonardBang]

@lvyanquan Would you like to rebase to latest master branch as conflicts happens?

[lvyanquan]

Rebased to master.

[leonardBang]

Thanks @lvyanquan for the contribution and @yuxiqian for the review, LGTM