12.10. 配置 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.3.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 集合结构和事件路由。要自定义行为，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 连接器在数据库更改时发出的更改事件消息外，连接器还会发出其他类型的信息，包括心跳消息，以及有关 schema 更改和事务的元数据消息。由于这些其他消息的结构与 SMT 设计的更改事件消息的结构不同，因此最好将连接器配置为有选择地应用 SMT，以便它只处理预期的数据更改消息。您可以使用以下任一方法配置连接器来有选择地应用 SMT：

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

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

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

默认情况下，payload 字段值( Avro 数据)是唯一消息值。配置 ByteArrayConverter，因为值转换器 将有效负载 字段值按原样传播到 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 集合可能会包含您要添加到发出出出出出消息的值的字段。例如，假设 outbox 集合的 aggregatetype 字段中的值为 purchase-order，另一个字段 eventType，其可能的值是 order-created 和 order-shipped。可以使用语法 field: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

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

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

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