FlinkCDC作业处理慢追不上binlog产生的速度，导致binlog被清理的解决方案_one or more fetchers have encountered exception

完整的异常信息如下：

java.lang.RuntimeException: One or more fetchers have encountered exception
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:225)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:185)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:143)
	at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:385)
	at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:526)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:811)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:760)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:954)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:933)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:568)
	at java.lang.Thread.run(Thread.java:834)
Caused by: 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:1147)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:622)
	... 1 more
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=1668149633760,db=,server_id=0,file=mysql-bin.003889,pos=116719908,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:190)
	at com.ververica.cdc.connectors.mysql.debezium.task.context.StatefulTaskContext.configure(StatefulTaskContext.java:116)
	at com.ververica.cdc.connectors.mysql.debezium.reader.BinlogSplitReader.submitSplit(BinlogSplitReader.java:99)
	at com.ververica.cdc.connectors.mysql.debezium.reader.BinlogSplitReader.submitSplit(BinlogSplitReader.java:67)
	at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.checkSplitOrStartNext(MySqlSplitReader.java:159)
	at com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader.fetch(MySqlSplitReader.java:71)
	at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58)
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:142)
	... 6 more
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

我的需求是FlinkCDC MySQL-AnalyticDB实时入库。

该报错的原因是作业处理的速度追不上binlog产生的速度，导致正在读的位点被清理了。目前只能清状态重启。

解决方案：

场景1:rds做了内部迁移操作，flink jar作业使用mysql cdc消费mysql数据报错的原因是是作业处理的速度追不上mysql binlog 产生的速度，导致正在读的位点被清理了
通过命令show master status，可以查看master数据库当前正在使用的二进制日志及当前执行二进制日志位置 show master logs,查看所有二进制日志列表
排查思路：
这种得先确认一下问题：rds的地址和binlog文件位点和迁移前是一致的吗？
1. rds 地址不会变化
2. 如果节点有变化，binlog文件位点会变化。如果是从远程拉取的oss中的binlog，binlog是固定的。
解决方案：
1：那需要重新读去了，flink cdc不会去oss上拉文件，是直链mysql服务器（无状态重启）
2：绕行方案：从只读库拉取数据 
注意事项：
（1）从库读是支持的，从库就是延时比主库大点。注意一点事 RDS MySQL 5.6不支持，5.7之后的版本都支持，因为RDS MySQL 5.6 只读实例的binlog文件是简化过的，没有数据。
（2）主库风险也还好，flink cdc 只有读的权限，不会加锁和写的权限。
如果rds 配置了HA，即多主实例，用户同时开启了GTID，然后通过 VIP/DNS 下挂rds的这几个多主实例地址，这样flink cdc /canal 这些同步工具 通过访问 VIP/DNS 链接rds时才能实现 不中断。

场景2:
RDS有日志保留策略，最长18个小时，最大占用30%存储空间，两个条件谁先满足都会触发删除，如果您写入特别多，超过30%的存储空间了，可能binlog日志1小时就删除了
注：rds页面上还有一个7天的binlog文件保存，这个是rds后台转存到您们的oss上的，flink cdc目前是没有去转存后oss上去读取这些文件的

场景3：
 volvo 通过只读实例消费 CDC 数据，RDS的只读实例不保证binlog（本地只保留10s，上传oss），所以 flink cdc 侧不建议连接 RDS 的只读实例。
只读实例一旦作业 Failover 10s 内恢复不过来，就会有这个异常
只读实例判定，rr 开头的就是只读实例 rm 开头的就是正常的实例
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22