Debezium日常分享系列之：Debezium 3.0.0.Beta发布

本文主要是介绍Debezium日常分享系列之：Debezium 3.0.0.Beta发布，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Debezium日常分享系列之：Debezium 3.0.0.Beta发布

一、重大变更
- Debezium Server Kafka Sink
- Debezium Server RabbitMQ Sink
二、新功能和改进
- 每个表的详细指标
- PostgreSQL复制插槽创建超时
- 对于PostgreSQL的PgVector数据类型的支持
- Oracle Ehcache 事务缓冲实现
- 解码PostgreSQL逻辑消息的转换
三、更多内容

此版本包含大量新功能和改进，包括每个表的创建、更新和删除的详细指标、复制槽创建超时、PostgreSQL 对 PgVector 数据类型的支持、基于 Ehcache 的新 Oracle 嵌入式缓冲区实现等。

一、重大变更

Debezium Server Kafka Sink

当Kafka代理不可用时，Debezium Server Kafka sink适配器可能会无限期等待。现在，在适配器中添加了一个可配置的超时选项，当达到超时时间时，适配器将强制失败。新的选项debezium.sink.kafka.wait.message.delivery.timeout.ms的默认值为30秒。如果默认值对您的需求不足，请相应地进行调整。

Debezium Server RabbitMQ Sink

Debezium Server RabbitMQ sink适配器将所有变更发送到同一个流中。虽然这在某些场景下可能很有用，但它与其他代理系统不太匹配，其他代理系统中每个表都会被流式传输到自己独特的主题或流中。从Debezium 3开始，这个逻辑已经发生了变化，默认情况下，每个表都将被流式传输到自己独特的流中。当设置debezium.sink.rabbitmqstream.stream时，您可以启用将所有变更流式传输到同一个流中的传统行为。

二、新功能和改进

每个表的详细指标

Debezium 现在将开始基于每个关系表执行的单个创建、更新和删除操作跟踪指标。对于某些连接器，如 PostgreSQL 和 Oracle，这些新的详细指标还会跟踪每个关系表执行的截断操作。这在需要检测特定变异模式或希望集成分析或可观测性堆栈的情况下非常有用，这些详细信息可以有助于识别问题。
对于升级到 Debezium 3 的用户，这些新指标将自动捕获。它们使用基于映射的模式 Map<‍String, Long‍> 暴露，其中键是表名，值是观察到的事件数量。新的指标名称是 NumberOfCreateEventsSeen、NumberOfDeleteEventsSeen、NumberOfUpdateEventsSeen 和 NumberOfTruncateEventsSeen。

PostgreSQL复制插槽创建超时

当首次部署PostgreSQL连接器时，其中一个最早的任务是在数据库中创建一个复制插槽，如果该插槽不存在。复制插槽是连接器工作的关键，它有助于捕获和发送更改到Debezium。不幸的是，有一些数据库操作会阻塞复制插槽的创建，比如正在进行的事务，强制连接器无限期地阻塞，等待事务结束。对于短暂的事务，这通常不是一个问题；然而，对于长时间运行的事务，情况就完全不同了。
为了改善这种体验，添加了一个新的内部选项
internal.create.slot.command.timeout，默认为90秒。如果复制插槽的创建在90秒内没有完成，它将重试slot.max.retries次。一旦重试次数用尽，连接器将抛出一个无法恢复的错误。

对于PostgreSQL的PgVector数据类型的支持

pgvector扩展为PostgreSQL引入了矢量搜索功能。该扩展引入了三种数据类型：vector、halfvec和sparsevec。
在Debezium 3中，所有三种数据类型都将像其他数据类型一样进行流式传输。
每种数据类型的发出基于以下语义映射：
- vector作为数值值的数组
- halfvec作为数值值的数组
- sparsevec作为具有维度数量和索引到值的映射的结构体
- 在启用数据库中的pgvector扩展后，无需进行其他配置。

Oracle Ehcache 事务缓冲实现

Debezium 3引入了基于Ehcache的新的Oracle连接器事务缓冲实现，以提供事务处理和事件数据的堆外存储。这个新的实现增加了现有的Java Heap、Infinispan Embedded和Infinispan Remote缓冲类型。
要开始利用Ehcache实现，必须将log.mining.buffer.type设置为ehcache。默认情况下，缓冲类型是memory，以使用JVM的堆获得最佳性能。
为了使Ehcache库成功启动，必须提供几个附加配置来明确配置缓存管理器维护的缓存。这些新的配置选项为：
- log.mining.buffer.ehcache.global.config
- log.mining.buffer.ehcache.transactions.config
- log.mining.buffer.ehcache.processedtransactions.config
- log.mining.buffer.ehcache.schemachanges.config
- log.mining.buffer.ehcache.events.config
Debezium使用XML创建Ehcache配置，因此这些配置提供XML片段。全局配置是可选的，并允许您提供有关持久性和其他Ehcache属性的详细信息，但不包括指定<‍cache‍>或<‍default-serializers‍>标记，这些标记是单独处理的。其他单个缓存配置旨在提供<‍cache‍>配置标记的内部XML位，但不包括其<‍key-type‍>和<‍value-type‍>，这些由Debezium直接管理。

配置示例

{"log.mining.buffer.type": "ehcache","log.mining.buffer.ehcache.global.config": "<persistence directory=\"./data\"/>","log.mining.buffer.ehcache.transactions.config": "<resources><heap unit=\"entries\">256</heap><disk unit=\"B\">10485760</disk></resources>","log.mining.buffer.ehcache.processedtransactions.config": "<resources><heap unit=\"entries\">256</heap><disk unit=\"B\">10485760</disk></resources>","log.mining.buffer.ehcache.schemachanges.config": "<resources><heap unit=\"entries\">256</heap><disk unit=\"B\">10485760</disk></resources>","log.mining.buffer.ehcache.events.config": "<resources><heap unit=\"entries\">256</heap><disk unit=\"B\">10485760</disk></resources>"
}

在这个示例中，Ehcache将为缓存维护堆和堆外存储的组合，在堆中始终保持最多256个条目，并刷新到磁盘。磁盘缓存将存储在相对路径./data下。这意味着在使用基于磁盘的缓存时，您需要一个可用的持久存储卷。

这是一个新的功能，属于实验性质，因此我们希望您能提供反馈意见，以便我们改进。

解码PostgreSQL逻辑消息的转换

PostgreSQL在于您可以通过使用pg_logical_emit_message将逻辑消息直接写入WAL来实现Outbox模式，而无需创建一个outbox表。不幸的是，这些数据随后以一系列字节的形式发送到Kafka，这对于可能正在寻找结构化消息的消费者可能并不总是理想的。
Debezium 3引入了一个新的针对PostgreSQL的特定转换，称为DecodeLogicalDecodingMessageContent。该转换的目的是将pg_logical_emit_message事件字节转换为消费者应用程序能够理解的结构化事件负载。

给定以下配置：

{"transforms": "decode","transforms.decode.type": "io.debezium.connector.postgresql.transforms.DecodeLogicalDecodingMessageContent"
}

在转换之前使用 pg_logic_emit_message 写入的事件的事件值将是：

{"op": "m","ts_ms": 1723115240065,"source": {...},"message": {"prefix": "test-prefix","content": "eyJpZCI6IDEsICJpdGVtIjogIkRlYmV6aXVtIGluIEFjdGlvbiIsICJzdGF0dXMiOiAiRU5URVJFRCIsICJxdWFudGl0eSI6IDIsICJ0b3RhbFByaWNlIjogMzkuOTh9"}
}

应用转换后，事件的值现在如下所示：

{"op": "c","ts_ms": 1723115415729,"source": {...},"after": {"id": 1,"item": "Debezium in Action","status": "ENTERED","quantity": 2,"totalPrice": 39.98}
}