在意外 AWS RDS binlog 删除后恢复 Debezium MySQL 连接器

Question

当 Debezium 作为 kafka 连接中的源运行时，如果目标 MySQL DB（Amazon RDS 实例）上一段时间内没有发生更新，那么一段时间后我会遇到以下错误。

[2018-04-25 21:30:14,526] INFO Step 0: Get all known binlogs from MySQL (io.debezium.connector.mysql.MySqlConnectorTask:310)
[2018-04-25 21:30:14,536] INFO Connector requires binlog file 'mysql-bin-changelog.002640', but MySQL only has mysql-bin-changelog.002663, mysql-bin-changelog.002664, mysql-bin-changelog.002665 (io.debezium.connector.mysql.MySqlConnectorTask:323)
[2018-04-25 21:30:14,536] INFO MySQL has the binlog file 'mysql-bin-changelog.002640' required by the connector (io.debezium.connector.mysql.MySqlConnectorTask:325)
[2018-04-25 21:30:14,536] INFO Stopping MySQL connector task (io.debezium.connector.mysql.MySqlConnectorTask:239)
[2018-04-25 21:30:14,536] INFO WorkerSourceTask{id=swiggy-connector-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:328)
[2018-04-25 21:30:14,536] INFO WorkerSourceTask{id=swiggy-connector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:345)
[2018-04-25 21:30:14,536] ERROR WorkerSourceTask{id=swiggy-connector-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:172)
org.apache.kafka.connect.errors.ConnectException: The connector is trying to read binlog starting at binlog file 'mysql-bin-changelog.002640', pos=470, skipping 4 events plus 0 rows, but this is no longer available on the server. Reconfigure the connector to use a snapshot when needed.
    at io.debezium.connector.mysql.MySqlConnectorTask.start(MySqlConnectorTask.java:117)
    at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:45)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:164)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
    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:748)

当我去DB检查MySQL中的binlog时

mysql> show binary logs;
+----------------------------+-----------+
| Log_name                   | File_size |
+----------------------------+-----------+
| mysql-bin-changelog.002664 |       479 |
| mysql-bin-changelog.002665 |       120 |
+----------------------------+-----------+


mysql> show binlog events;
+----------------------------+-----+-------------+------------+-------------+---------------------------------------------------------------------------------------------------------------------------------+
| Log_name                   | Pos | Event_type  | Server_id  | End_log_pos | Info                                                                                                                            |
+----------------------------+-----+-------------+------------+-------------+---------------------------------------------------------------------------------------------------------------------------------+
| mysql-bin-changelog.002664 |   4 | Format_desc | 1550192458 |         120 | Server ver: 5.6.39-log, Binlog ver: 4                                                                                           |
| mysql-bin-changelog.002664 | 120 | Query       | 1550192458 |         201 | BEGIN                                                                                                                           |
| mysql-bin-changelog.002664 | 201 | Query       | 1550192458 |         391 | use `mysql`; INSERT INTO mysql.rds_heartbeat2(id, value) values (1,1524671965007) ON DUPLICATE KEY UPDATE value = 1524671965007 |
| mysql-bin-changelog.002664 | 391 | Xid         | 1550192458 |         422 | COMMIT /* xid=308462 */                                                                                                         |
| mysql-bin-changelog.002664 | 422 | Rotate      | 1550192458 |         479 | mysql-bin-changelog.002665;pos=4                                                                                                |
+----------------------------+-----+-------------+------------+-------------+---------------------------------------------------------------------------------------------------------------------------------+

问题：

1. Debezium为什么闲着？为什么它没有从 MySQL 读取 002640 文件之后的文件？
   任何服务均未使用此功能。因此，在 Debezium 可以读取数据之前，不可能出现太多写入操作。
2. 当没有发生任何活动时，为什么Amazon MySQL RDS会删除 binlog 文件？
   这是一个测试数据库，只有我在其中插入记录。所以这里没有发生外部应用程序活动。
3. 有没有办法恢复Debezium连接器并开始处理 MySQL 当前可用的时间日志中的记录？（如果我同意那些未读记录丢失的话）。
   我尝试重新启动作业，删除并添加连接器，但总是出现相同的错误。
   唯一适用于恢复活动的解决方案
   • 删除Kafka Connect的offet主题。
   • 删除并再次添加 debezium 连接器。
     我想要一种不同的方法，因为在生产中我们将有大量的连接器将使用相同的偏移主题。所以删除是不可能的。

Answer 1

请查看heartbeat.interval.ms配置属性 - 这应该可以防止在高流量环境中由 Debezium 监控低流量表的情况。在这种情况下，可能会发生 binlog 已刷新但当前 binlog 坐标未记录在 offsets 主题中的情况。

关于简历 - 您可以通过修改 offsets 主题来恢复这种情况。在这里，您需要插入服务器上可用的插件和二进制日志坐标的偏移记录。有一个Kafka KIP工具可以帮助解决这个问题。现在你需要手动完成。