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

要了解 Debezium outbox 事件路由器 SMT 是如何配置的，请查看以下 Debezium outbox 消息示例：

# Kafka Topic: outbox.event.order
# Kafka Message key: "1"
# Kafka Message Headers: "id=4d47e190-0402-4048-bc2c-89dd54343cdc"
# Kafka Message Timestamp: 1556890294484
{
  "{\"id\": 1, \"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 事件路由器 SMT 的 Debezium 连接器通过转换 Debezium 原始消息来生成上述消息，如下所示：

# Kafka Message key: "406c07f3-26f0-4eea-a50c-109940064b8f"
# Kafka Message Headers: ""
# Kafka Message Timestamp: 1556890294484
{
  "before": null,
  "after": {
    "id": "406c07f3-26f0-4eea-a50c-109940064b8f",
    "aggregateid": "1",
    "aggregatetype": "Order",
    "payload": "{\"id\": 1, \"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}",
    "timestamp": 1556890294344,
    "type": "OrderCreated"
  },
  "source": {
    "version": "2.3.4.Final",
    "connector": "postgresql",
    "name": "dbserver1-bare",
    "db": "orderdb",
    "ts_usec": 1556890294448870,
    "txId": 584,
    "lsn": 24064704,
    "schema": "inventory",
    "table": "outboxevent",
    "snapshot": false,
    "last_snapshot_record": null,
    "xmin": null
  },
  "op": "c",
  "ts_ms": 1556890294484
}

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

要应用 default outbox 事件路由器 SMT 配置，您的 outbox 表被认为具有以下列：

Column        |          Type          | Modifiers
--------------+------------------------+-----------
id            | uuid                   | not null
aggregatetype | character varying(255) | not null
aggregateid   | character varying(255) | not null
type          | character varying(255) | not null
payload       | jsonb                  |

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

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter

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

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

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

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
value.converter=io.debezium.converters.BinaryDataConverter

默认情况下，有效负载 列值( Avro 数据)是唯一的消息值。配置 BinaryDataConverter，因为值转换器将 payload 列值按原样传播到 Kafka 消息值中。

Debezium 连接器可以配置为发送心跳、事务元数据或模式更改事件（支持因连接器而异）。这些事件无法通过 BinaryDataConverter 序列化，因此必须提供额外的配置，因此转换器知道如何序列化这些事件。例如，以下配置演示了使用没有模式的 Apache Kafka JsonConverter ：

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
value.converter=io.debezium.converters.BinaryDataConverter
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 列中具有 购买顺序 的值，另一个列, eventType，其可能的值是 order-created 和 order-shipped。可以使用语法 column:placement:alias 来添加其他字段。

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

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

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

结果将是 Kafka 消息的标头，类型 作为其键，eventType 列的值作为其值。

要在 outbox 消息 envelope 中发出 eventType 列值，请配置 SMT，如下所示：

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

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

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.table.fields.additional.placement=partitionColumn:partition

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

您可能注意到，Debezium outbox 消息包含以 String 表示 的有效负载。因此，当这个字符串实际上是 JSON 时，它会出现在结果 Kafka 信息中，如下所示：

# Kafka Topic: outbox.event.order
# Kafka Message key: "1"
# Kafka Message Headers: "id=4d47e190-0402-4048-bc2c-89dd54343cdc"
# Kafka Message Timestamp: 1556890294484
{
  "{\"id\": 1, \"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 事件路由器允许您将此消息内容扩展为 "real" JSON，使用从 JSON 文档本身分离的编译模式。这样，Kafka 信息的结果类似如下：

# Kafka Topic: outbox.event.order
# Kafka Message key: "1"
# Kafka Message Headers: "id=4d47e190-0402-4048-bc2c-89dd54343cdc"
# Kafka Message Timestamp: 1556890294484
{
  "id": 1, "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
}

要启用此转换，您必须将 table.expand.json.payload 设置为 true，并使用如下所示的 JsonConverter ：

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.table.expand.json.payload=true
value.converter=org.apache.kafka.connect.json.JsonConverter

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