12.3. 根据事件内容将更改事件记录路由到主题

为安全起见，Debezium 连接器存档不包含基于内容的路由 SMT。相反，它会在单独的工件中提供 debezium-scripting-2.1.4.Final.tar.gz。

如果通过从 Dockerfile 构建自定义 Kafka Connect 容器镜像来部署 Debezium 连接器，要使用过滤器 SMT，您必须将 SMT 工件明确添加到 Kafka Connect 环境中。当使用 AMQ Streams 部署连接器时，它可以根据您在 Kafka Connect 自定义资源中指定的配置参数自动下载所需的工件。重要信息：在 Kafka Connect 实例中存在路由 SMT 后，允许向实例添加连接器的任何用户都可以运行脚本表达式。要确保脚本表达式只能由授权用户运行，请务必在添加路由 SMT 前保护 Kafka Connect 实例及其配置接口。

如果您从 Dockerfile 构建 Kafka Connect 容器镜像，则适用以下步骤。如果使用 AMQ Streams 创建 Kafka Connect 镜像，请按照连接器部署主题中的说明操作。

Groovy 语言在 classpath 上需要以下库：

JavaScript 语言在 classpath 上需要以下库：

要将 Debezium 连接器配置为根据事件内容路由更改事件记录，您可以在连接器的 Kafka Connect 配置中配置 ContentBasedRouter SMT。

配置基于内容的路由 SMT 要求您指定一个定义过滤条件的正则表达式。在配置中，您可以创建一个定义路由条件的正则表达式。表达式定义了用于评估事件记录的模式。它还指定目的地主题的名称，其中路由与模式匹配的事件。您指定的模式可能会指定一个事件类型，如表插入、更新或删除操作。您还可以定义一个与特定列或行中值匹配的模式。

例如，要将所有更新(u)记录重新路由到 updates 主题，您可以在连接器配置中添加以下配置：

...
transforms=route
transforms.route.type=io.debezium.transforms.ContentBasedRouter
transforms.route.language=jsr223.groovy
transforms.route.topic.expression=value.op == 'u' ? 'updates' : null
...

...
transforms=route
transforms.route.type=io.debezium.transforms.ContentBasedRouter
transforms.route.language=jsr223.groovy
transforms.route.topic.expression=value.op == 'u' ? 'updates' : null
...

前面的例子指定了使用 Groovy 表达式语言。

与模式不匹配的记录将路由到默认主题。

Debezium 将某些变量绑定到 SMT 的评估上下文中。当您创建表达式来指定控制路由目的地的条件时，SMT 可以查找并解释这些变量的值，以评估表达式中的条件。

下表列出了 Debezium 绑定到基于内容路由 SMT 的评估上下文的变量：

表达式可以在其变量上调用任意方法。表达式应解析为布尔值，用来决定 SMT 如何处理消息。当表达式中的路由条件评估为 true 时，会保留消息。当路由条件评估为 false 时，会删除消息。

表达式不应产生任何副作用。也就是说，不应修改他们通过的任何变量。

除了 Debezium 连接器在发生数据库更改时发出的更改事件消息外，连接器还会发出其他类型的消息，包括心跳消息以及有关架构更改和事务的元数据消息。由于这些消息的结构与 SMT 旨在处理的更改事件消息的结构不同，因此最好将连接器配置为有选择地应用 SMT，以便它只处理预期的数据更改消息。您可以使用以下方法之一将连接器配置为有选择地应用 SMT：

您表达基于内容的路由条件的方式取决于您使用的脚本语言。例如，如 基本配置示例所示，当您使用 Groovy 作为表达式语言时，以下表达式会将所有更新(u)记录重新路由到 更新 主题，而将其他记录路由到默认主题：

value.op == 'u' ? 'updates' : null

value.op == 'u' ? 'updates' : null

其他语言使用不同的方法表达相同的条件。

value.get('op') == 'u' ? 'updates' : null

value.get('op') == 'u' ? 'updates' : null

value.op == 'u' ? 'updates' : null

value.op == 'u' ? 'updates' : null