问题标签 [debezium]

我正在开发一个连接到逻辑复制槽的应用程序，以使用 WAL 事件。然后将这些 WAL 事件转发到 MQ 代理。这很好用，但我注意到一段时间后我的内存不足。我设法将问题最小化为负责获取 WAL 事件的代码。它发生在以下代码中：

我注释掉了解析消息并将其转发到 MQ 代理的逻辑，因为没有这个也会发生内存不足。

我还尝试通过使用轮询方法readPending()而不是阻塞方法来更改此示例read()（如https://jdbc.postgresql.org/documentation/head/replication.html所示），但问题仍然存在。

我还注意到，过了一会儿，应用程序的 CPU 使用率达到了 100%。这一定是由底层库引起的，因为此时read()仍在正常处理（也就是说，它按顺序处理每个 WAL 事件）。

在这些测试期间，我以低速率执行INSERT和查询。UPDATE

我正在使用以下依赖项：

该应用程序在 Tomcat8 容器中作为 WAR 运行。

知道发生了什么吗？

更新 1

我弄清楚发生了什么，但到目前为止无法解释。我会详细说。

如前所述，我每隔 10 秒就会进行一次INSERT查询UPDATE。这些查询导致 645 个 WAL 事件。所以每 10 秒，我就有read()645 个事件。一开始，read()一个事件需要 0（或有时 1）毫秒。一段时间后，需要 1 毫秒。然后，再过一段时间，它需要 2 毫秒。等等...

所以过了一段时间，我无法read()在 10 秒内完成 645 个事件，因为所需的时间read()不断增加。这解释了 100% 的 CPU 使用率和内存不足。

我仍然不确定如何解释以及如何解决这个问题。我会继续调查。

更新 2

我尝试buffer.clear()在循环结束时添加，但没有成功。我仍然遇到 100% CPU 和内存问题。正如预期的那样，因为缓冲区是一个局部变量，所以它在每次循环之后都会以任何方式进行 GC。但我认为无论如何测试都是个好主意。

我已经在 Kafka 连接中集成了 debezium 连接器来流式传输我的 mysql 数据库更改。对于某些程序，它无法读取 binlog 文件。有什么建议或解决方法吗？卡住了这个问题。我可以在 mysql 中指定不要将 ddl 日志放入 binlog 文件吗？

我正在使用 Kafka Connect 的 Debezium 插件来流式传输 MySQL 数据库更改。

我在_

连接器属性

table.whitelist=tripDriverMapping

database.tables=azuga.tripDriverMapping

为什么 Debezium 为数据库中的所有表创建主题？是否有任何解决方法可以避免创建所有这些不必要的主题，因为我将只使用一个主题。

我已经阅读了http://debezium.io/docs/connectors/mysql/但我找不到任何关于是否可以配置 debezium 的信息，以便可以将来自 2 个（或更多）表的更改写入相同的单个 kafka话题？在我看来，它总是 1 个表 -> 1 个主题。

我正在尝试使用Python解释Debezium在Kafka中存储的Avro记录

我不确定这对应于哪个 Python 3 原始类型。这个值如何反序列化？

提前致谢！

我有spring boot + hibernate应用程序，现在我想使用 debezium 来捕获日期变化。假设这个类是我的 POJO 实体类之一：

现在，如果我创建新Article对象并使用如下代码将其持久化：

然后再次加载并更改一些字段并将其持久化：

然后删除它：

我可能会在每个步骤中看到这样的日志：

正如您在上面看到的，我无法检测到如何从数据库中删除记录。我怎么解决这个问题？我应该使用像Hibernate Enver这样的东西 来补充这个吗？

有人使用 Debezium kafka 测试集群进行 gradle 单元测试吗？

我有一个 Vertx gradle 项目，并希望将 debezium 测试集群与我的 vertcl kafka 客户端测试集成。

我的第一个问题是进口，它们似乎不起作用。如何转换 maven 样式测试导入？<type>test-jar</type>

vertx kafka 客户端有一个示例测试，所以我希望一旦我克服了 gradle 导入它就可以工作。

这不足以获得正确的进口

这个测试库是否记录在某处？

提前致谢

G。

作为 MySQL 的“替代品”，我相信相同的 binlog 和复制语义，我假设在 Debezium 文档中的任何地方都说“mysql”，我可以用“percona”（或就此而言 MariaDB）替换它，但我想在我浪费很多时间或遇到意外的陷阱或错误之前进行确认。

我对 KafkaStreams 上的密钥反序列化有疑问。具体来说，我使用 Kafka Connect 和 debezium 连接器从 Postgres 表中读取数据。数据被导入到 Kafka 主题中，在 Kafka 模式注册表上创建了两个 Avro 模式，一个用于键，一个用于值（这包含表上的所有列）。

我在 GlobalKTable 上读取了这些数据，如下所示：

我的问题是我有一个拓扑，我需要将这个 GlobalKTable 与一个 KStream 连接起来，如下所示：

请注意，my.namespace.db.Key 的 Avro 架构是

显然 GlobalKTable 和 KStream 上的键是不同的对象，我不知道如何实现连接。我最初尝试过这个，但没有奏效。

我得到的输出如下，我可以在我的一个连接主题上看到一个警告（这似乎很可疑），但没有其他任何连接实体的输出，就好像没有什么可消耗的。

我可以让这个加入在 Kafka Streams 上工作吗？请注意，如果我使用 KTable 读取主题并在 KStream 上使用 selectKey 来转换密钥，则此方法有效，但我想避免重新分区。
或者正确的方法是否应该以另一种方式从数据库中导入我的数据以避免创建 Avro 对象，以及如何使用 debezium 连接器和启用 AvroConverter 的 KafkaConnect 来实现？

使用 Debezium 0.7 从 MySQL 读取，但在初始快照阶段出现刷新超时和 OutOfMemoryError 错误。查看下面的日志，似乎连接器试图一次写入太多消息：

想知道对于大型数据库 (>50GB)的正确设置是什么http://debezium.io/docs/connectors/mysql/#connector-properties 。对于较小的数据库，我没有这个问题。简单地增加超时似乎不是一个好策略。我目前正在使用默认连接器设置。

更新

按照以下建议更改了设置并解决了问题：