Debezium系列之：Debezium 通知
- 一、概述
- 二、Debezium 通知格式
- 三、可用的通知
- 四、启用 Debezium 通知
- 五、访问 Debezium JMX 通知
- 六、自定义通知渠道
- 七、配置自定义通知渠道
- 八、Debezium 核心模块依赖项
- 九、部署自定义通知渠道
- 十、配置连接器以使用自定义通知通道
一、概述
Debezium 通知提供了一种获取有关连接器状态信息的机制。通知可以发送到以下渠道：
- 接收器通知通道：通过 Connect API 将通知发送到配置的主题。
- 日志通知通道：通知会附加到日志中。
- JmxNotificationChannel：通知作为 JMX bean 中的属性公开。
- Custom：通知将发送到您实施的自定义渠道。
二、Debezium 通知格式
通知消息包含以下信息：
| 属性 | 描述 |
|---|---|
| id | 分配给通知的唯一标识符。对于增量快照通知，该 ID 与使用执行快照信号发送的 ID 相同。 |
| aggregate_type | 与通知相关的聚合根的数据类型。在领域驱动设计中，导出的事件应始终引用聚合。 |
| type | 提供有关在aggregate_type 字段中指定的事件的状态信息。 |
| additional_data | 包含有关通知的详细信息的 Map<String,String>。有关示例，请参阅有关增量快照进度的 Debezium 通知。 |
三、可用的通知
Debezium 通知提供有关初始快照或增量快照进度的信息。
初始快照的状态
以下示例显示了提供初始快照状态的典型通知：
{
"id": "5563ae14-49f8-4579-9641-c1bbc2d76f99",
"aggregate_type": "Initial Snapshot",
"type": "COMPLETED"
}
类型字段可以包含以下值之一：
- COMPLETED
- ABORTED
- SKIPPED
Debezium 有关增量快照进度的通知
下表显示了报告增量快照状态的通知中可能存在的不同负载的示例：
Start：
{
"id":"ff81ba59-15ea-42ae-b5d0-4d74f1f4038f",
"aggregate_type":"Incremental Snapshot",
"type":"STARTED",
"additional_data":{
"connector_name":"my-connector",
"data_collections":"table1, table2"
}
}
Paused：
{
"id":"068d07a5-d16b-4c4a-b95f-8ad061a69d51",
"aggregate_type":"Incremental Snapshot",
"type":"PAUSED",
"additional_data":{
"connector_name":"my-connector",
"data_collections":"table1, table2"
}
}
Resumed：
{
"id":"a9468204-769d-430f-96d2-b0933d4839f3",
"aggregate_type":"Incremental Snapshot",
"type":"RESUMED",
"additional_data":{
"connector_name":"my-connector",
"data_collections":"table1, table2"
}
}
Stopped：
{
"id":"83fb3d6c-190b-4e40-96eb-f8f427bf482c",
"aggregate_type":"Incremental Snapshot",
"type":"ABORTED",
"additional_data":{
"connector_name":"my-connector"
}
}
Processing chunk：
{
"id":"d02047d6-377f-4a21-a4e9-cb6e817cf744",
"aggregate_type":"Incremental Snapshot",
"type":"IN_PROGRESS",
"additional_data":{
"connector_name":"my-connector",
"data_collections":"table1, table2",
"current_collection_in_progress":"table1",
"maximum_key":"100",
"last_processed_key":"50"
}
}
Snapshot completed for a table：
{
"id":"6d82a3ec-ba86-4b36-9168-7423b0dd5c1d",
"aggregate_type":"Incremental Snapshot",
"type":"TABLE_SCAN_COMPLETED",
"additional_data":{
"connector_name":"my-connector",
"data_collection":"table1",
"total_rows_scanned":"100",
"status":"SUCCEEDED"
}
}
可能的值为：
- EMPTY - 表为空
- NO_PRIMARY_KEY - 表没有快照所需的主键
- SKIPPED - 不支持此类表的快照，请检查日志以了解详细信息
- SQL_EXCEPTION - 处理快照时捕获 SQL 异常
- SUCCEEDED - 快照成功完成
- UNKNOWN_SCHEMA - 找不到表的架构，请检查日志以获取已知表的列表
Completed：
{
"id":"6d82a3ec-ba86-4b36-9168-7423b0dd5c1d",
"aggregate_type":"Incremental Snapshot",
"type":"COMPLETED",
"additional_data":{
"connector_name":"my-connector"
}
}
四、启用 Debezium 通知
要使 Debezium 能够发出通知，请通过设置 notification.enabled.channels 配置属性来指定通知通道列表。默认情况下，以下通知渠道可用：
- sink
- log
- jmx
要使用接收器通知通道，您还必须将 notification.sink.topic.name 配置属性设置为您希望 Debezium 发送通知的主题的名称。
五、访问 Debezium JMX 通知
要使 Debezium 能够报告通过 JMX beans 公开的事件，请完成以下配置步骤：
- 启用 JMX MBean 服务器以公开通知 bean。
- 将 jmx 添加到连接器配置中的 notification.enabled.channels 属性中。
- 将您首选的 JMX 客户端连接到 MBean 服务器。
通知通过名称为 debezium.<connector-type>.management.notifications.<server> 的 bean 的“Notifications”属性公开。
下图显示了报告增量快照开始的通知：
要放弃通知，请对 bean 调用重置操作。
通知还公开为 debezium.notification 类型的 JMX 通知。要使应用程序能够侦听 MBean 发出的 JMX 通知，请为应用程序订阅通知。
六、自定义通知渠道
通知机制被设计为可扩展的。您可以根据需要实施渠道，以最适合您的环境的方式传递通知。添加通知通道涉及几个步骤：
- 为通道创建一个Java项目来实现通道，并添加Debezium Core作为依赖项。
- 部署通知通道。
- 通过修改连接器配置，使连接器能够使用自定义通知通道。
七、配置自定义通知渠道
自定义通知通道是实现 io.debezium.pipeline.notification.channels.NotificationChannel 服务提供者接口 (SPI) 的 Java 类。例如：
public interface NotificationChannel {
String name();
void init(CommonConnectorConfig config);
void send(Notification notification);
void close();
}
- 频道的名称。要使 Debezium 能够使用该通道，请在连接器的 notification.enabled.channels 属性中指定此名称。
- 初始化通道所需的特定配置、变量或连接。
- 在频道上发送通知。 Debezium 调用此方法来报告其状态。
- 关闭所有分配的资源。 Debezium 在连接器停止时调用此方法。
八、Debezium 核心模块依赖项
自定义通知通道 Java 项目具有对 Debezium 核心模块的编译依赖项。您必须将这些编译依赖项包含在项目的 pom.xml 文件中，如以下示例所示：
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-core</artifactId>
<version>${version.debezium}</version>
</dependency>
${version.debezium} 表示 Debezium 连接器的版本。
在 META-INF/services/io.debezium.pipeline.notification.channels.NotificationChannel 文件中声明您的实现。
九、部署自定义通知渠道
先决条件：
- 您有一个自定义通知通道 Java 程序。
程序
- 要将通知通道与 Debezium 连接器结合使用，请将 Java 项目导出到 JAR 文件，然后将该文件复制到包含要与其一起使用的每个 Debezium 连接器的 JAR 文件的目录。
- 例如，在典型部署中，Debezium 连接器文件存储在 Kafka Connect 目录 (/kafka/connect) 的子目录中，每个连接器 JAR 位于其自己的子目录中 (/kafka/connect/debezium-connector-db2、/kafka /connect/debezium-connector-mysql 等）。要将信号通道与连接器一起使用，请将转换器 JAR 文件添加到连接器的子目录中。
十、配置连接器以使用自定义通知通道
将自定义通知通道的名称添加到 notification.enabled.channels 配置属性中。
