How can i catch flink-cdc exception and handle it manually？

[lshuoz]

1. After i submit my flink-cdc job， how can i catch the source exception and handle it manually？
   2023-12-18 17:16:53,675 ERROR [blc-10.87.15.89:3316] [io.debezium.connector.mysql.MySqlStreamingChangeEventSource] - Error during binlog processing. Last offset stored = null, binlog reader near position = sdp-mysql-bin.000014/11932 2023-12-18 17:16:53,676 ERROR [blc-10.87.15.89:3316] [io.debezium.pipeline.ErrorHandler] - Producer failure io.debezium.DebeziumException: Could not find first log file name in binary log index file Error code: 1236; SQLSTATE: HY000. at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1383) at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1439) at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:980) at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:599) at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:857) at java.lang.Thread.run(Thread.java:750) Caused by: com.github.shyiko.mysql.binlog.network.ServerException: Could not find first log file name in binary log index file at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:944) ... 3 more 2023-12-18 17:16:54,000 ERROR [Source Data Fetcher for Source: Mysql Source -> Sink: Unnamed (1/1)#186] [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) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:105) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:750) Caused by: org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped. at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:42) at com.ververica.cdc.connectors.mysql.debezium.task.context.MySqlErrorHandler.setProducerThrowable(MySqlErrorHandler.java:89) at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1439) at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:980) at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:599) at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:857) ... 1 more Caused by: io.debezium.DebeziumException: Could not find first log file name in binary log index file Error code: 1236; SQLSTATE: HY000. at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1383) ... 5 more Caused by: com.github.shyiko.mysql.binlog.network.ServerException: Could not find first log file name in binary log index file at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:944) ... 3 more

[lshuoz]

I hava submit my job like：
`
MySqlSource mySqlSource = MySqlSource.builder()
.hostname(mysqlConfig.getHost())
.port(mysqlConfig.getPort())
.databaseList(mysqlConfig.getDatabaseList()) // set captured database
.username(mysqlConfig.getUsername())
.password(mysqlConfig.getPassword())
.tableList(mysqlConfig.getTableList()) // set captured table
// .fetchSize(1000)
.includeSchemaChanges(false)
.startupOptions(StartupOptions.initial())
.debeziumProperties(props)
.deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
.scanNewlyAddedTableEnabled(true)
.build();

    Configuration conf = new Configuration();
    conf.setString(RestOptions.BIND_PORT, "9071");

    StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

// StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // enable checkpoint
    env.enableCheckpointing(3000);
    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    env.setStateBackend(new FsStateBackend("file:///F:/checkpoint/checkpints"));
    env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "Mysql Source")
            .addSink(new AssetsSink())
            .setParallelism(1); // use parallelism 1

    JobExecutionResult jobResult = null;
    jobResult = env.execute("Print MySQL Snapshot + Binlog");

`