12.8. 配置 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.1.4.Final",
    "connector": "mongodb",
    "name": "fulfillment",
    "ts_ms": 1558965508000,
    "snapshot": false,
    "db": "inventory",
    "rs": "rs0",
    "collection": "customers",
    "ord": 31,
    "h": 1546547425148721999
  },
  "op": "c",
  "ts_ms": 1556890294484
}

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

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

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

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

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

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

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

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

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

默认情况下，有效负载 字段值( Avro 数据)是唯一消息值。将 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。

您的 outbox 集合可能会包含您要添加到发出的消息中的值。例如，假设在 aggregatetype 字段中具有 product -order 值的 outbox 集合，另一个字段 eventType，其可能的值是 order-created 和 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 message 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

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

默认情况下，Debebe 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

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