[Feature]: Support switch from Changestore to Logstore automatically for flink source

[zhoujinsong]

Description

Support switch from Changestore to Logstore automatically for flink source

Use case/motivation

Arctic use Changestore and Logstore to store CDC messages.
Logstore use MQ(eg: Kafka) to distribute real-time cdc message.
Changestore use iceberg table to store cdc message for analyse and provide near real-time incremental pull.
In really use case, Logstore store message for several days and Changestore store message for a longer time.
Flink source should support to read Changestore for early CDC messages and switch to Logstore when catch it up.

Describe the solution

• Support to config the reading start time for Flink source
• Flink source determin read Changestore or Logstore according reading start time
• Flink source should transfer from Changestore to Logstore when catch it up.

Subtasks

• [Improvement][Flink]: Refactor Log-store Source to FLINK FLIP-27 API #969

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct