在意外删除 AWS RDS 二进制日志后恢复 Debezium MySQL 连接器

Posted

技术标签:

【中文标题】在意外删除 AWS RDS 二进制日志后恢复 Debezium MySQL 连接器【英文标题】：Resume Debezium MySQL connector after unexpected AWS RDS binlogs deletion 【发布时间】：2018-10-06 05:01:07 【问题描述】：

当 Debezium 作为 kafka connect 中的源运行时，如果该目标 mysql 数据库（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 WorkerSourceTaskid=swiggy-connector-0 Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:328)
[2018-04-25 21:30:14,536] INFO WorkerSourceTaskid=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 WorkerSourceTaskid=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 中的 binlogs 时

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                                                                                                |
+----------------------------+-----+-------------+------------+-------------+---------------------------------------------------------------------------------------------------------------------------------+

问题：

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

【问题讨论】：

我不知道 #1 或 #3 但是 RDS 会在不需要日志时立即删除它们对于 RDS 知道的任何事情（例如停滞的 RDS 副本)，这通常意味着在正常条件下大约 5 到 10 分钟内。您无法让 RDS 了解外部连接器，但可以以 1 小时为增量延迟此自动日志过期，总共最多 7 天：docs.aws.amazon.com/AmazonRDS/latest/UserGuide/… @Michael-sqlbot 谢谢。我现在可以将其用作解决方法。将其增加到一天会导致任何性能问题吗？我计划在 1 小时到 1 天之间更改此配置，具体取决于数据库的空闲程度。如果我继续这样做，我应该担心吗？ 【参考方案1】：

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

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

【讨论】：

我尝试配置 heartbeat.interval.ms 但仍然没有帮助。当我在连接器中将几个表列入白名单并且这些表没有发生更新时，我的偏移量不会在 debezium 中增加。为了恢复这个过程，我手动尝试更新偏移量，但它仍然没有帮助。 bin/kafka-console-producer.sh --broker-list localhost:9092 --topic cdc-offsets 与数据 "file":"mysql-bin-changelog.003188","pos":120,"event":2。我应该修改哪个主题来更新最后读取的 binlog 文件名和偏移量？唯一可行的方法是删除主题 cdc-offsets 并重新启动连接。 你使用什么 Debezium 版本？ 我使用的是 0.7.5 你找到解决这个问题的办法了吗？ 偏移操作已记录在 debezium.io/docs/faq/…

以上是关于在意外删除 AWS RDS 二进制日志后恢复 Debezium MySQL 连接器的主要内容，如果未能解决你的问题，请参考以下文章

如果AWS RDS恢复发生，如何重新连接

sh 将AWS RDS二进制文件备份到AWS S3

AWS：为单独的 EB 实例克隆 RDS

AWS BeanStalk 不倾向于连接到 AWS RDS

如何在 AWS elasticbeanstalk 上删除环境而不删除 RDS

误删mysql数据后，如何通过mysql二进制日志恢复数据