问题标签 [debezium]

有人能告诉我使用带有 osgi 包的嵌入式 Debezium（没有 kafka）和使用带有 kafka 的 Debezium 有什么区别吗？我在互联网上找不到很多关于 Debezium 的信息。

我在 apache karaf 上使用 osgi，我正在尝试使用kafka和debezium来运行osgi 环境。

kafka和debezium还没有准备好 osgi（karaf 不会将它们视为捆绑包），所以我确实使用 eclipse“插件项目”对它们进行了 osgified。我对它们进行 osgified 的 jar 如下：debezium-embedded、debezium-core、kafka connect-api、kafka connect-runtime。

一开始，当我尝试运行 debezium 时，我得到了很多“找不到类的异常”。

为了解决这个问题，我更改了两个捆绑包的清单。我向调用者添加了一个导入包，向被调用包添加了一个导出包。使用它我可以解决 classNotFound 问题。

解决所有 classNotfound 问题后，我得到NoClassDefFoundError

NoClassDefFoundError 意味着类加载器在尝试加载它们时找不到 .class ......但我确实导入了所有包并导出它们。

任何想法如何在osgi 环境中处理NoClassDefFoundError

[编辑添加代码]

这是类 Monitor ：

还有我的激活器：

MySql ROW 级别的 binloging 已启用。

我正在使用 Confluent 平台 3.2。在 3 台不同的 EC2 机器上运行 3 名工作人员。

我有一个连接器（debezium/MySQL 源），我将其删除并在几分钟后重新启动。但是由于以下错误，我无法成功启动连接器。连接器处于故障状态。我不得不重新启动工人来解决问题。

需要知道这是否是缓存问题？如何在不重新启动工作人员的情况下解决此问题。任何支持表示赞赏。

I have problem of capturing data in mysql with debezium change data capture and consuming it to another mysql using kafka connect jdbc sink.

Because the schema and payload that debezium produces to kafka topic is not compatible with the schema that kafka connect jdbc sink expects.

I get exception when jdbc sink wants to consume data and create records in another mysql.

How should I solve this problem ?

I am using debezium mongodb connector 0.3.6 , running inside a docker container. I have been monitoring kafka-connect for some time ,and found that the connector stops tailing mongodb and sending change events to kafka brokers automatically. Upon investigation , I found that sometimes, after some time of inactivity, it's mongo connection is refused, and upon retrial, it connects successfully ,and sends the large number of records that it has not sent during inactive period. But this is not the case everytime, as I have encountered this inactivity for 1 hour straight, and upon restarting the connector deliberately, I can get it to tail the oplog properly, and hence sends an enormous number of records that were pending during inactivity.

I am attaching 2 log files: - 1- consisting logs where the mongo connection is refused, and it automatically retry and connects successfully. (https://ufile.io/c5v5g) 2- consisting logs where for 1 hour, it was not tailing mongodb oplog, and no change events were sent to kafka , even when new events were written in mongodb oplog. (https://ufile.io/9g4gz)

I am not able to find a particular problem here. My kafka-connect container, mongodb host, and kafka brokers are all running on different hosts, and connected by public ip.

Debezium 连接器的 Kafka 连接事件是 Avro 编码的。

在传递给 Kafka 连接独立服务的 connect-standalone.properties 中提到了以下内容。

使用以下属性配置 Kafka 使用者代码：

在消费者实现中，以下是读取键和值组件的代码。我正在使用 REST 从模式注册表中获取键和值的模式。

解析密钥工作正常。在解析消息的值部分时，我得到了 ArrayIndexOutOfBoundsException。

下载了 Avro 的源代码并进行了调试。发现GenericDatumReader.readInt方法返回的是负值。这个值应该是数组（符号）的索引，因此应该是正数。

尝试使用 kafka-avro-standalone-consumer 消费事件，但它也抛出了 ArrayIndexOutOfBoundsException。所以，我的猜测是消息在 Kafka 连接（生产者）处编码不正确，问题出在配置上。

以下是问题：

1. 生产者或消费者传递的配置有什么问题吗？
2. 为什么密钥反序列化有效但价值无效？
3. 是否需要做其他事情才能使事情正常进行？（比如在某处指定字符编码）。
4. 带有 Avro 的 Debezium 可以在生产中使用，还是现在是一个实验性功能？Debezium Avro 上的帖子特别指出，未来将包含涉及 Avro 的示例。

有很多帖子 Avro 反序列化抛出 ArrayIndexOutOfBoundsException 但无法将其与我面临的问题联系起来。

提供了一个用例：

流处理架构；事件进入 Kafka，然后由具有 MongoDB 接收器的作业处理。

数据库名称：myWebsite 集合：users

user并且作业在users集合中下沉记录。

1. 所以 Debezium 将监视users集合的变化，并且在每次变化时，都会在 Kafka 中产生有关该主题的事件dbserver1.myWebsite.users？假设dbserver1是连接器的名称。
2. 如果是这样，那么我可以有一个 Kafka 消费者来消费dbserver1.myWebsite.users主题并对这些事件做出反应？
3. 据我了解，Debezium 产生的事件也包含数据库记录的值？如果它的变化包含旧/新值？如果创建了一个数据库记录，旧的是否为空？

到目前为止，我想对我的理解进行某种确认。谢谢！

我们正在使用 Debezium + PostgreSQL。

请注意，我们获得了 4 种类型的创建、读取、更新和删除事件 - c、r、u 和 d。

我们的应用程序没有使用读取类型的事件。实际上，除非我们正在审计或镜像事务的活动，否则我无法想到“r”事件的用例。

我们面临扩展困难，我怀疑这是因为网络被读取类型的事件占用。

我们如何在 postgreSQL 本身中过滤掉这些事件？

我从其中一位贡献者那里得到了使用 snapshot.mode 的线索。我想当 Debezium 创建快照时必须做一些事情。我无法弄清楚如何做到这一点。