背景
Debezium引擎在进行数据同步的时候，如果期间发生了异常问题。会有专门的异常处理类来出来，MySQL异常处理类如下：io.debezium.connector.mysql.MySqlErrorHandler.java
isRetriable方法作用：如果识别到指定的异常信息，引擎会触发定时重启。而对于网络异常，目前1.6.0.Final版本没法识别到，模拟网络端口后，会有如下的异常：
2021-07-27 17:56:26.481 ERROR 8556 --- [30.233.147:3306] io.debezium.pipeline.ErrorHandler : Producer failure
io.debezium.DebeziumException: Network is unreachable: recv failed
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1154) ~[debezium-connector-mysql-1.6.0.Final.jar:1.6.0.Final]
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1199) [debezium-connector-mysql-1.6.0.Final.jar:1.6.0.Final]
at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:973) [mysql-binlog-connector-java-0.25.1.jar:0.25.1]
at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606) [mysql-binlog-connector-java-0.25.1.jar:0.25.1]
at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850) [mysql-binlog-connector-java-0.25.1.jar:0.25.1]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_271]
Caused by: java.net.SocketException: Network is unreachable: recv failed
at java.net.SocketInputStream.socketRead0(Native Method) ~[na:1.8.0_271]
at java.net.SocketInputStream.socketRead(SocketInputStream.java:116) ~[na:1.8.0_271]
at java.net.SocketInputStream.read(SocketInputStream.java:171) ~[na:1.8.0_271]
at java.net.SocketInputStream.read(SocketInputStream.java:141) ~[na:1.8.0_271]
at com.github.shyiko.mysql.binlog.io.BufferedSocketInputStream.read(BufferedSocketInputStream.java:51) ~[mysql-binlog-connector-java-0.25.1.jar:0.25.1]
at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.readWithinBlockBoundaries(ByteArrayInputStream.java:222) ~[mysql-binlog-connector-java-0.25.1.jar:0.25.1]
at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.peek(ByteArrayInputStream.java:194) ~[mysql-binlog-connector-java-0.25.1.jar:0.25.1]
at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:930) [mysql-binlog-connector-java-0.25.1.jar:0.25.1]
... 3 common frames omitted
当网络恢复的时候，引擎无法正常工作。需要将此异常添加到ErrorHandler类中。
新建异常处理类
在工程项目，手动创建io.debezium.connector.mysql.MySqlErrorHandler类，将SocketException加上去。代码如下：
public class MySqlErrorHandler extends ErrorHandler {
private static final String SQL_CODE_TOO_MANY_CONNECTIONS = "08004";
public MySqlErrorHandler(String logicalName, ChangeEventQueue<?> queue) {
super(MySqlConnector.class, logicalName, queue);
}
@Override
protected boolean isRetriable(Throwable throwable) {
if (throwable instanceof SQLException) {
final SQLException sql = (SQLException) throwable;
return SQL_CODE_TOO_MANY_CONNECTIONS.equals(sql.getSQLState());
}
else if (throwable instanceof ServerException) {
final ServerException sql = (ServerException) throwable;
return SQL_CODE_TOO_MANY_CONNECTIONS.equals(sql.getSqlState());
}
else if(throwable instanceof SocketException) {
return true;
}
else if (throwable instanceof DebeziumException && throwable.getCause() != null) {
return isRetriable(throwable.getCause());
}
return false;
}
}
这样修改后，启动项目工程。然后模拟网络异常问题，则控制台会打印出重连信息，如下：
但是没有一直打印下去，所以猜测使用的默认配置，需要手动进行配置指定等待时长。
访问github debezium代码库，找到打印的相关类：io.debezium.connector.common.BaseSourceTask.java，阅读代码很容易找到读取配置的地方：
于是在引擎启动的地方，配置此参数。比如设置为半小时。配置是在CommonConnectorConfig.java类中，默认配置为10s。
启动的时候，设置重试等待时间为半小时：
重新启动引擎
然后模拟网络异常问题，发现一直打印消息，需要说明的是，其内部就是个do while循环，循环到指定时间开始重启引擎，可以理解为等待多长时间重启引擎，此机制不能立马感知网络恢复，只是做了个定时重启而已。
网友评论