[INLONG-7828][Sort] Mysql CDC check the availability of binlog based on gitd firstly

[e-mhui]

Prepare a Pull Request

[INLONG-7828][Sort] Mysql CDC check the availability of binlog based on gitd firstly

• Fixes [Improve][Sort] MySQL CDC reads Binlog files from GTID #7828

Motivation

When connecting to the Mysql master, if the master-slave switch is executed, the binlog file of the master will change, and the task may fail to find the binlog file.

Caused by: java.lang.IllegalStateException: The connector is trying to read binlog starting at Struct{version=1.5.4.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1681704265482,db=,server_id=0,file=mysql-bin.000014,pos=315,row=0}, but this is no longer available on the server. Reconfigure the connector to use a snapshot when needed.
	at org.apache.inlong.sort.cdc.mysql.debezium.task.context.StatefulTaskContext.loadStartingOffsetState(StatefulTaskContext.java:187) ~[classes/:?]
	at org.apache.inlong.sort.cdc.mysql.debezium.task.context.StatefulTaskContext.configure(StatefulTaskContext.java:119) ~[classes/:?]
	at org.apache.inlong.sort.cdc.mysql.debezium.reader.BinlogSplitReader.submitSplit(BinlogSplitReader.java:90) ~[classes/:?]
	at org.apache.inlong.sort.cdc.mysql.debezium.reader.BinlogSplitReader.submitSplit(BinlogSplitReader.java:61) ~[classes/:?]
	at org.apache.inlong.sort.cdc.mysql.source.reader.MySqlSplitReader.checkSplitOrStartNext(MySqlSplitReader.java:165) ~[classes/:?]
	at org.apache.inlong.sort.cdc.mysql.source.reader.MySqlSplitReader.fetch(MySqlSplitReader.java:74) ~[classes/:?]
	at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:56) ~[flink-connector-base-1.13.5.jar:1.13.5]
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:140) ~[flink-connector-base-1.13.5.jar:1.13.5]
	... 6 more

Referring to this apache/flink-cdc#761, we need to first check binlog availability based on GTID, and then check binlog availability from binlog files.

Modifications

Check the availability of binlog based on gitd firstly.

    private boolean isBinlogAvailable(MySqlOffsetContext offset) {
        String gtidSet = offset.gtidSet();
        if (gtidSet != null) {
            return checkGtidSet(offset);
        }
        return checkBinlogFilename(offset);