Outbox Event Router

This single message transformation (SMT) is available since Debezium version 0.9.3. It is under active development right now, so the emitted message structure or other details may still change as development progresses.

This SMT does not support the MongoDB connector.

The Outbox Event Router SMT has the intent to provide out-of-the-box support for the Outbox Pattern. For more information on why and how the Outbox Pattern is used please refer our blog post Reliable Microservices Data Exchange With the Outbox Pattern.

For more implementation details please refer to the Outbox SMT Design.

A working example can also be found at our Examples Repository in the outbox directory.

Example

In order to understand the configuration and terms used in this SMT, let’s look into its parts with the given expected outbox message:

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

This message is generated by transforming a Debezium raw message, which looks like:

# 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": "0.9.3.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
}

This result was achieved with the default configuration which assumes a table structure and event routing based on aggregates. In case you want a custom behavior, the SMT is fully configurable, check the available configuration options.

Configuration

Configuration options

Property Default Group Description

debezium.op.invalid.behavior

warn

Debezium

While Debezium is monitoring the table, it’s not expecting to see 'update' row events, in case it happens, this transform can log it as warning, error or stop the process. Options are warn, error and fatal

table.field.event.id

id

Table

The column which contains the event ID within the outbox table

table.field.event.key

aggregateid

Table

The column which contains the event key within the outbox table; when this is set the value of this column will be used as a Kafka message key

table.field.event.type

type

Table

The column which contains the event type within the outbox table

table.field.event.timestamp

Table

Optionally you can override the Kafka message timestamp with a value from a chosen field, otherwise it’ll be the Debezium event processed timestamp.

table.field.event.payload

payload

Table

The column which contains the event payload within the outbox table

table.field.event.payload.id

aggregateid

Table

The column which contains the payload ID within the outbox table

table.fields.additional.placement

Table, Envelope

Extra fields can be added as part of the event envelope or a message header; the format is a list of colon-delimited pairs or trios when you desire to have aliases, e.g. id:header,field_name:envelope:alias. Placement options are header and envelope

table.field.event.schema.version

Table, Schema

When set, it’ll be used as schema version as in the Kafka Connect Schema javadoc

route.by.field

aggregatetype

Router

The column which determines how the events will be routed, the value will become part of the topic name

route.topic.regex

(?<routedByValue>.*)

Router

The default regex to use within the RegexRouter, the default capture will allow to replace the routed field into a new topic name defined in route.topic.replacement

route.topic.replacement

outbox.event.${routedByValue}

Router

The name of the topic in which the events will be routed, a replacement ${routedByValue} is available which is the value of The column configured via route.by.field

Default table columns

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                  | not null

Default configuration values explained

After observing all those pieces we can see what the default configuration does:

Table Column Effect

payload

The JSON representation of the event itself, becomes part of the message as payload

id

The id shows up as a header in the Kafka message, this is the unique ID of the event, can be used for consumer side dedupe for instance.

aggregatetype

Is the default field for the routing, it gets append to the topic name (check configuration route.topic.replacement)

aggregateid

Becomes the Kafka message key, which is important for keeping ordering within Kafka partitions

type

The type column becomes part of the message envelope as eventType

Basic configuration

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
back to top