Search before asking

• I searched in the issues and found nothing similar.

Flink version

1.16.0

Flink CDC version

2.3.0

Database and its version

mysql5.7

Minimal reproduce step

if i restart job from checkpoint can not work

2023-02-21 14:08:29,713 INFO com.ververica.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext [] - MySQL current GTID set 106a4bb6-ec0d-11ec-a2d4-00163e279211:1-204182899,7aec1281-719c-11eb-afcf-00163e06a35c:1-147359662 does contain the GTID set 106a4bb6-ec0d-11ec-a2d4-00163e279211:203495054-204182173 required by the connector.
2023-02-21 14:08:29,801 INFO com.ververica.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext [] - Server has already purged 106a4bb6-ec0d-11ec-a2d4-00163e279211:1-203495053,7aec1281-719c-11eb-afcf-00163e06a35c:1-147359662 GTIDs
2023-02-21 14:08:29,802 WARN com.ververica.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext [] - Some of the GTIDs needed to replicate have been already purged
2023-02-21 14:08:29,803 ERROR org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager [] - Received uncaught exception.
java.lang.RuntimeException: SplitFetcher thread 0 received unexpected exception while polling the records
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:150) ~[flink-connector-files-1.16.1.jar:1.16.1]
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:105) [flink-connector-files-1.16.1.jar:1.16.1]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_362]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_362]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_362]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_362]
at java.lang.Thread.run(Thread.java:750) [?:1.8.0_362]
Caused by: java.lang.IllegalStateException: The connector is trying to read binlog starting at Struct{version=1.6.4.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1676959709803,db=,server_id=0,file=mysql-bin.005893,pos=15052069,row=0}, but this is no longer available on the server. Reconfigure the connector to use a snapshot when needed.
at com.ververica.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext.loadStartingOffsetState(StatefulTaskContext.java:194) ~[blob_p-384c1d850f997d73acba83fc72dc8cfb8bead162-07d02e9eebcf14d5793c784b1d89c6cc:?]
at com.ververica.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext.configure(StatefulTaskContext.java:117) ~[blob_p-384c1d850f997d73acba83fc72dc8cfb8bead162-07d02e9eebcf14d5793c784b1d89c6cc:?]
at com.ververica.cdc.connectors.mysql.debezium.reader.BinlogSplitReader.submitSplit(BinlogSplitReader.java:103) ~[blob_p-384c1d850f997d73acba83fc72dc8cfb8bead162-07d02e9eebcf14d5793c784b1d89c6cc:?]
at com.ververica.cdc.connectors.mysql.debezium.reader.BinlogSplitReader.submitSplit(BinlogSplitReader.java:71) ~[blob_p-384c1d850f997d73acba83fc72dc8cfb8bead162-07d02e9eebcf14d5793c784b1d89c6cc:?]
at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.checkSplitOrStartNext(MySqlSplitReader.java:159) ~[blob_p-384c1d850f997d73acba83fc72dc8cfb8bead162-07d02e9eebcf14d5793c784b1d89c6cc:?]
at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.fetch(MySqlSplitReader.java:71) ~[blob_p-384c1d850f997d73acba83fc72dc8cfb8bead162-07d02e9eebcf14d5793c784b1d89c6cc:?]
at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58) ~[flink-connector-files-1.16.1.jar:1.16.1]
at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:142) ~[flink-connector-files-1.16.1.jar:1.16.1]
... 6 more

gtids in checkpoint:"file":"mysql-bin.005893","pos":"15052069","kind":"SPECIFIC","gtids":"106a4bb6-ec0d-11ec-a2d4-00163e279211:203495054-204182173"
show master status gtids :106a4bb6-ec0d-11ec-a2d4-00163e279211:1-204479617,
7aec1281-719c-11eb-afcf-00163e06a35c:1-147359662
show binary logs：

What did you expect to see?

if first set specificOffset then can work from ck

What did you see instead?

if first set specificOffset then can not work from ck

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!