7.10. 配置 Debezium MongoDB 连接器以使用 outbox 模式

注意

这个 SMT 仅用于 Debezium MongoDB 连接器。有关将 outbox 事件路由器 SMT 用于关系数据库的详情，请参考 Outbox 事件路由器。

outbox 模式是一种在多个(micro)服务之间安全且可靠地交换数据的方法。outbox 模式实现可避免服务内部状态（通常是在其数据库中保留）和需要相同数据的服务所消耗的事件之间的不一致。

要在 Debezium 应用程序中实施 outbox 模式，请将 Debezium 连接器配置为：

捕获 outbox 集合中的更改
应用 Debezium MongoDB outbox 事件路由器单一消息转换(SMT)

配置为应用 MongoDB outbox SMT 的 Debezium 连接器应该只捕获在 outbox 集合中发生的更改。如需更多信息，请参阅用于有选择地应用转换的选项。

只有在每个 outbox 集合具有相同的结构时，连接器才能捕获多个 outbox 集合中的更改。

注意

要使用这个 SMT，对实际业务集合的操作和插入到 outbox 集合的操作必须作为多文档事务的一部分完成，这些事务自 MongoDB 4.0 开始被支持，以防止商业集合和出箱集合间的潜在的数据不一致。对于将来的更新，要启用更新现有数据并在没有多文档事务的 ACID 事务中插入 outbox 事件，我们计划支持以现有集合的子文档的形式存储 outbox 事件，而不是独立开箱即用集合。

有关开箱即用模式的更多信息，请参阅使用 Outbox 模式进行可靠的发布数据交换。

以下主题提供详情：

7.10.1. Debezium MongoDB outbox 消息示例

要了解如何配置 Debezium MongoDB outbox 事件路由器 SMT，请考虑以下 Debezium outbox 消息示例：

# Kafka Topic: outbox.event.order
# Kafka Message key: "b2730779e1f596e275826f08"
# Kafka Message Headers: "id=596e275826f08b2730779e1f"
# Kafka Message Timestamp: 1556890294484
{
  "{\"id\": {\"$oid\": \"da8d6de63b7745ff8f4457db\"}, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}"
}

配置为应用 MongoDB outbox 事件路由器 SMT 的 Debezium 连接器，通过转换原始 Debezium 更改事件信息来生成前面的消息，如下例所示：

# Kafka Message key: { "id": "{\"$oid\": \"596e275826f08b2730779e1f\"}" }
# Kafka Message Headers: ""
# Kafka Message Timestamp: 1556890294484
{
  "patch": null,
  "after": "{\"_id\": {\"$oid\": \"596e275826f08b2730779e1f\"}, \"aggregateid\": {\"$oid\": \"b2730779e1f596e275826f08\"}, \"aggregatetype\": \"Order\", \"type\": \"OrderCreated\", \"payload\": {\"_id\": {\"$oid\": \"da8d6de63b7745ff8f4457db\"}, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}}",
  "source": {
    "version": "2.7.3.Final",
    "connector": "mongodb",
    "name": "fulfillment",
    "ts_ms": 1558965508000,
    "ts_us": 1558965508000000,
    "ts_ns": 1558965508000000000,
    "snapshot": false,
    "db": "inventory",
    "rs": "rs0",
    "collection": "customers",
    "ord": 31,
    "h": 1546547425148721999
  },
  "op": "c",
  "ts_ms": 1556890294484,
  "ts_us": 1556890294484452,
  "ts_ns": 1556890294484452697,
}

此 Debezium outbox 消息示例基于默认的 outbox 事件路由器配置，它假定 outbox 集合结构和事件路由基于聚合。要自定义行为，outbox 事件路由器 SMT 提供了大量配置选项。

7.10.2. Debezium mongodb outbox 事件路由器 SMT 期望的 outbox 集合结构

要应用默认的 MongoDB outbox 事件路由器 SMT 配置，您的 outbox 集合被认为有以下字段：

{
  "_id": "objectId",
  "aggregatetype": "string",
  "aggregateid": "objectId",
  "type": "string",
  "payload": "object"
}

表 7.10. expected outbox 集合字段的描述
字段	效果
`id`	包含事件的唯一 ID。在 outbox 消息中，这个值是一个标头。例如，您可以使用此 ID 删除重复的消息。要从不同的 outbox 集合字段获取事件的唯一 ID，请在连接器配置中设置 `collection.field.event.id` SMT 选项。
`aggregatetype`	包含一个值，它把 SMT 附加到主题名称中，连接器会发出 outbox 信息。默认行为是，这个值替换了 `route.topic.replacement` SMT 选项中的默认 `${routedByValue}` 变量。例如，在默认配置中，`route.by.field` SMT 选项被设置为 `aggregatetype`，`route.topic.replacement` SMT 选项被设置为 `outbox.event.${routedByValue}`。假设您的应用程序在 outbox 集合中添加了两个文档。在第一个文档中，`aggregatetype` 字段中的值是 `customer`。在第二文档中，`aggregatetype` 字段中的值是 `orders`。连接器将第一个文档发送到 `outbox.event.customers` 主题。连接器将第二个文档发送到 `outbox.event.orders` 主题。要从不同的 outbox 集合字段获取这个值，请在连接器配置中设置 `route.by.field` SMT 选项。
`aggregateid`	包含事件键，它为有效负载提供 ID。SMT 使用这个值作为发出的 outbox 消息中的键。这对于在 Kafka 分区中维护正确顺序非常重要。要从不同的 outbox 集合字段获取事件密钥，请在连接器配置中设置 `collection.field.event.key` SMT 选项。
`payload`	outbox 更改事件表示。默认结构是 JSON。默认情况下，Kafka 消息值仅由 `payload` 值组成。但是，如果 outbox 事件配置为包含其他字段，则 Kafka 消息值包含接收两个有效负载和其他字段，每个字段将单独表示。如需更多信息，请参阅使用其他字段传输消息。要从不同的 outbox 集合字段获取事件有效负载，请在连接器配置中设置 `collection.field.event.payload` SMT 选项。
其它自定义字段	outbox 集合中的任何其他字段都可以添加到 payload 部分或作为消息标头中的 outbox 事件中。一个示例可以是字段 `eventType`，它传递了一个用户定义的值，有助于分类或组织事件。

7.10.3. 基本 Debezium MongoDB outbox 事件路由器 SMT 配置

要将 Debezium MongoDB 连接器配置为支持 outbox 模式，请配置 outbox.MongoEventRouter SMT。要获取 SMT 的默认行为，请将其添加到连接器配置中，而不指定任何选项，如下例所示：

transforms=outbox,...
transforms.outbox.type=io.debezium.connector.mongodb.transforms.outbox.MongoEventRouter

自定义配置

连接器可能会发出许多类型的事件消息（如 heartbeat 消息、tombstone 消息或有关事务的元数据消息）。要将转换仅应用到来自 outbox 集合的事件，请定义一个 SMT predicate 语句，该语句仅有选择地将转换应用到这些事件。

7.10.4. 用于有选择地应用 MongoDB Outbox 事件路由器转换的选项

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

为转换配置 SMT predicate。
为 SMT 使用 route.topic.regex 配置选项。

7.10.5. 在 Debezium MongoDB outbox 信息中使用 Avro 作为有效负载格式

MongoDB outbox 事件路由器 SMT 支持任意有效负载格式。outbox 集合中的 payload 字段值会透明地传递。使用 JSON 的替代方法是使用 Avro。这对消息格式监管非常有用，并确保 outbox 事件模式以向后兼容的方式进行演进。

源应用程序如何为 outbox 消息有效负载生成 Avro 格式的内容，而不是本文档的范围。一个可能是利用 KafkaAvroSerializer 类来序列化 GenericRecord 实例。要确保 Kafka 消息值是确切的 Avro 二进制数据，请将以下配置应用到连接器：

transforms=outbox,...
transforms.outbox.type=io.debezium.connector.mongodb.transforms.outbox.MongoEventRouter
value.converter=io.debezium.converters.ByteArrayConverter

默认情况下，payload 字段值( Avro data)是唯一的消息值。配置 ByteArrayConverter 作为值转换器将 payload 字段值按原样传播到 Kafka 消息值。

请注意，这与为其他 SMT 建议的 BinaryDataConverter 不同。这是因为 MongoDB 在内部存储字节阵列的不同方法。

Debezium 连接器可以被配置为发出心跳、事务元数据或架构更改事件（支持因连接器而异）。这些事件不能由 ByteArrayConverter 序列化，因此必须提供额外的配置，以便转换器知道如何序列化这些事件。例如，以下配置演示了使用带有任何模式的 Apache Kafka JsonConverter ：

transforms=outbox,...
transforms.outbox.type=io.debezium.connector.mongodb.transforms.outbox.MongoEventRouter
value.converter=io.debezium.converters.ByteArrayConverter
value.converter.delegate.converter.type=org.apache.kafka.connect.json.JsonConverter
value.converter.delegate.converter.type.schemas.enable=false

delegate Converter 实现由 delegate.converter.type 选项指定。如果转换器需要任何额外的配置选项，也可以指定它们，如使用 schemas.enable=false 所示的模式禁用。

7.10.6. 在 Debezium MongoDB outbox 信息中发出其他字段

您的 outbox 集合可能包含您要添加到发出的 outbox 消息中的值。例如，考虑在 aggregatetype 字段中具有 purchase-order 值的 outbox 集合，另一个字段 eventType，其可能的值是 order-create 和 order-shipped。可以使用语法 字段：placement:alias 添加其他字段。

放置 允许的值有： - header - envelope - partition

要在 outbox 消息标头中发出 eventType 字段值，请配置 SMT，如下所示：

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.collection.fields.additional.placement=eventType:header:type

结果将是 Kafka 消息中的一个标头，类型为 type 作为其键，以及 eventType 字段的值作为其值。

要发送 outbox 消息 envelope 中的 eventType 字段值，请配置 SMT，如下所示：

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.collection.fields.additional.placement=eventType:envelope:type

要控制生成 outbox 消息的分区，请配置 SMT，如下所示：

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.collection.fields.additional.placement=partitionField:partition

请注意，对于 分区 放置，添加一个别名无效。

7.10.7. 扩展转义的 JSON String 作为 JSON

默认情况下，Debezium outbox 消息 有效负载 以字符串表示。当字符串的原始源采用 JSON 格式时，生成的 Kafka 消息使用转义序列来代表字符串，如下例所示：

# Kafka Topic: outbox.event.order
# Kafka Message key: "1"
# Kafka Message Headers: "id=596e275826f08b2730779e1f"
# Kafka Message Timestamp: 1556890294484
{
  "{\"id\": {\"$oid\": \"da8d6de63b7745ff8f4457db\"}, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}"
}

您可以配置 outbox 事件路由器来扩展消息内容，将转义的 JSON 转换为其原始的未转义的 JSON 格式。在转换的字符串中，companion 模式会从原始 JSON 文档中删除。以下示例显示了生成的 Kafka 消息中扩展的 JSON：

# Kafka Topic: outbox.event.order
# Kafka Message key: "1"
# Kafka Message Headers: "id=596e275826f08b2730779e1f"
# Kafka Message Timestamp: 1556890294484
{
  "id": "da8d6de63b7745ff8f4457db", "lineItems": [{"id": 1, "item": "Debezium in Action", "status": "ENTERED", "quantity": 2, "totalPrice": 39.98}, {"id": 2, "item": "Debezium for Dummies", "status": "ENTERED", "quantity": 1, "totalPrice": 29.99}], "orderDate": "2019-01-31T12:13:01", "customerId": 123
}

要在转换中启用字符串转换，请将 collection.expand.json.payload 的值设置为 true，并使用 StringConverter，如下例所示：

transforms=outbox,...
transforms.outbox.type=io.debezium.connector.mongodb.transforms.outbox.MongoEventRouter
transforms.outbox.collection.expand.json.payload=true
value.converter=org.apache.kafka.connect.storage.StringConverter

7.10.8. 配置 outbox 事件路由器转换的选项

下表描述了您可以为 outbox 事件路由器 SMT 指定的选项。在表中，Group 列指示 Kafka 的配置选项分类。

表 7.11. outbox 事件路由器 SMT 配置选项的描述
选项	默认	组	描述
`collection.op.invalid.behavior`	`warn`	集合	决定在 outbox 集合上有更新操作时 SMT 的行为。可能的设置有： `warn` - SMT 会记录警告，并继续到下一个 outbox 集合文档。 `error` - SMT 记录错误，并继续到下一个 outbox 集合文档。 `fatal` - SMT 日志错误，连接器会停止处理。 outbox 集合中的所有更改都应该是一个插入或删除操作。也就是说，开箱即用的集合充当队列；不允许对 outbox 集合中文档的更新。SMT 会在 outbox 集合上自动过滤删除操作（用于删除继续的 outbox 事件）。
`collection.field.event.id`	`_id`	集合	指定包含唯一事件 ID 的 outbox 集合字段。此 ID 将存储在 `id` 键下的已发送事件的标头中。
`collection.field.event.key`	`aggregateid`	集合	指定包含事件键的 outbox 集合字段。当此字段包含值时，SMT 使用该值作为发出的 outbox 消息中的键。这对于在 Kafka 分区中维护正确顺序非常重要。
`collection.field.event.timestamp`		集合	默认情况下，发出的 outbox 消息中的时间戳是 Debezium 事件时间戳。要在 outbox 消息中使用不同的时间戳，请将此选项设置为 outbox 集合字段，其中包含您要发出的 outbox 消息的时间戳。
`collection.field.event.payload`	`payload`	集合	指定包含事件有效负载的 outbox 集合字段。
`collection.expand.json.payload`	`false`	集合	指定是否应执行 String 有效负载的 JSON 扩展。如果没有找到内容或解析错误，则内容将保持"as is"。如需了解更多详细信息，请参阅扩展转义的 json 部分。
`collection.fields.additional.placement`		集合, Envelope	指定您要添加到 outbox 消息标头或 envelopes 的一个或多个 outbox 集合字段。指定以逗号分隔的对列表。在每个对中，指定字段的名称以及是否希望该值位于标头中还是信封中。使用冒号将对中的值分开，例如： `id:header,my-field:envelope` 要为字段指定别名，请将别名的 trio 指定为第三个值，例如： `id:header,my-field:envelope:my-alias` 第二个值是放置，它始终必须是 `header` 或 `envelope`。配置示例包括在 emitting additional fields in Debezium outbox messages 中。
`collection.field.event.schema.version`		collection, Schema	设置后，这个值将用作 schema 版本，如 Kafka Connect Schema Javadoc 所述。
`route.by.field`	`aggregatetype`	路由器	指定 outbox 集合中字段的名称。默认情况下，此字段中指定的值成为连接器向发送 outbox 消息的主题名称的一部分。例如，请参阅预期 outbox 集合的描述。
`route.topic.regex`	`(?<routedByValue>.*)`	路由器	指定在 RegexRouter 中将 outbox SMT 应用到 outbox 集合的正则表达式。此正则表达式是 `route.topic.replacement` SMT 选项的设置的一部分。当此属性设置为默认的 `Router` 值时，SMT 会将 `route.topic.replacement` 属性中设置的默认 `${routedByValue}` 变量替换为为 `route.by.field` 属性指定的值。
`route.topic.replacement`	`outbox.event.${routedByValue}`	路由器	指定连接器发出 outbox 消息的主题名称。默认主题名称以字符串 `outbox.event.` 前缀，后跟分配给 outbox 集合文档中的 `aggregatetype` 字段的值。例如，如果 `aggregatetype` 值是 customer，则连接器将 outbox 消息发送到主题 `outbox.event. customers` 。要更改目标主题的默认名称，请执行以下操作之一：将 `route.by.field` 选项设置为不同的字段。将 `route.topic.regex` 选项设置为不同的正则表达式。
`route.tombstone.on.empty.payload`	`false`	路由器	指明是否为空或 `null` 有效负载会导致连接器发出 tombstone 事件。