[FLINK-37120][pipeline-connector/mysql] Add ending split chunk first to avoid TaskManager oom

[beryllw]

Add ending split chunk first to avoid TaskManager oom

[beryllw]

CI：https://github.com/beryllw/flink-cdc/actions/runs/12779452404
@leonardBang Could you please help me retrigger the CI?

[beryllw]

Adding the ending split chunk first changes the snapshot read result order. I will fix the unit test.

[leonardBang]

THanks @beryllw for the contribution, the idea makes sense to me, but we need to consider the compatibility, a configuration is recommended

[beryllw]

THanks @beryllw for the contribution, the idea makes sense to me, but we need to consider the compatibility, a configuration is recommended

If data consistency is not an issue, are there any compatibility concerns we need to address?

[lvyanquan]

What‘s more, could you add this optimization to MongoDBChunkSplitter?

[beryllw]

What‘s more, could you add this optimization to MongoDBChunkSplitter?

Sure, i will check MongoDBChunkSplitter.

[beryllw]

What‘s more, could you add this optimization to MongoDBChunkSplitter?

#3704 (comment)

I overlooked this issue #3704, good idea. Maybe support AssignStrategy in base module is better? Do we implement this in a future PR or in this PR?

[lvyanquan]

I missed this issue #3704, good idea. Maybe support AssignStrategy in base module is better? Do we implement this in a future PR or in this PR?

My previous PR provided both ascending_order and descending_order order configurations, but I think it would be better to place both start and end unbounded chunks at the beginning, so you can go ahead.

I think if there is no impact on restarting from the previous state, there is no compatibility issue, but adding a parameter to control it would be safer.

[lvyanquan]

I think adding a test in MySqlSourceITCase to test the case of restoring from failure will be better.

[beryllw]

I think adding a test in MySqlSourceITCase to test the case of restoring from failure will be better.

agree.

[lvyanquan]

LGTM.
I'm worried if this parameter is a bit complicated, what about naming it scan.incremental.assign-max-chunk-first.enabled to avoid using ending? As chunk is a concept in snapshot phase.
WDYT @beryllw @leonardBang?

[beryllw]

LGTM. I'm worried if this parameter is a bit complicated, what about naming it scan.incremental.assign-max-chunk-first.enabled to avoid using ending? As chunk is a concept in snapshot phase. WDYT @beryllw @leonardBang?

What about scan.incremental.snapshot.assign-ending-chunk-first.enabled, when the write throughput is low, the ending chunk is not equal to the maximum chunk.

[beryllw]

@leonardBang @lvyanquan PTAL

[lvyanquan]

Thanks @beryllw for this update, since you've added this feature to jdbc common module, it seems like that this config option was not added in DynamicTableSourceFactory and was not passed to Source.

[beryllw]

Thanks @beryllw for this update, since you've added this feature to jdbc common module, it seems like that this config option was not added in DynamicTableSourceFactory and was not passed to Source.

Thanks for pointing out the error, fixed. Awaiting CI.
https://github.com/beryllw/flink-cdc/actions/runs/13256583939?pr=7

[lvyanquan]

Hi, @beryllw.
What I want to point out is that this option was not included in optionalOptions of PostgreSQLTableFactory, and was also not used in createDynamicTableSource, So it didn't take effect in SQL.

Do you want to support it all here, or do you want to split the modification of JDBC base into another PR.

[beryllw]

Do you want to support it all here, or do you want to split the modification of JDBC base into another PR.

Support it all here is better.Thanks.

[beryllw]

https://github.com/beryllw/flink-cdc/actions/runs/13258656371?pr=7
@leonardBang Please help me retrigger the CI.

[beryllw]

CI https://github.com/beryllw/flink-cdc/actions/runs/13276446389.