[id="outbox-event-router"]
= Outbox Event Router

:toc:
:toc-placement: macro
:linkattrs:
:icons: font
:source-highlighter: highlight.js

toc::[]

[NOTE]
====
This single message transformation (SMT) is under active development right now, so the emitted message structure or other details may still change as development progresses.
====

[NOTE]
====
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 link:/blog/2019/02/19/reliable-microservices-data-exchange-with-the-outbox-pattern/[Reliable Microservices Data Exchange With the Outbox Pattern].

For more implementation details please refer to the {jira-url}/browse/DBZ-1169[Outbox SMT Design].

A working example can also be found at our https://github.com/debezium/debezium-examples[Examples Repository] in the https://github.com/debezium/debezium-examples/tree/master/outbox[`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:

[source,javascript,indent=0]
----
# 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 {prodname} raw message, which looks like:

[source,javascript,indent=0]
----
# 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 {link-prefix}:{link-outbox-event-router}#outbox-event-router-configuration-options[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 {link-prefix}:{link-outbox-event-router}#outbox-event-router-configuration-options[available configuration options].

== Configuration

[[outbox-event-router-configuration-options]]
=== Configuration options
[cols="30%a,10%a,10%a,50%a"]
|===
|Property
|Default
|Group
|Description

|[[outbox-event-router-property-table-field-event-id]]<<outbox-event-router-property-table-field-event-id, `table.field.event.id`>>
|`id`
|Table
|The column which contains the event ID within the outbox table

|[[outbox-event-router-property-table-field-event-key]]<<outbox-event-router-property-table-field-event-key, `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

|[[outbox-event-router-table-field-event-type]]<<outbox-event-router-table-field-event-type, `table.field.event.type`>>
|`type`
|Table
|The column which contains the event type within the outbox table

|[[outbox-event-router-property-table-field-event-timestamp]]<<outbox-event-router-property-table-field-event-timestamp, `table.field.event.timestamp`>>
|
|Table
|Optionally you can override the Kafka message timestamp with a value from a chosen field, otherwise it is the {prodname} event processed timestamp.

|[[outbox-event-router-property-table-field-event-payload]]<<outbox-event-router-property-table-field-event-payload, `table.field.event.payload`>>
|`payload`
|Table
|The column which contains the event payload within the outbox table

|[[outbox-event-router-property-table-field-event-payload-id]]<<outbox-event-router-property-table-field-event-payload-id, `table.field.event.payload.id`>>
|`aggregateid`
|Table
|The column which contains the payload ID within the outbox table

|[[outbox-event-router-property-table-fields-additional-placement]]<<outbox-event-router-property-table-fields-additional-placement, `table.fields.additional.placement`>>
|
|Table, Envelope
|Extra fields can be added as part of the event envelope or as 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`. +
It is possible to configure `eventType` field among additional fields.
In such case the field could be placed either in envelope or among other headers.

|[[outbox-event-router-property-table-field-event-schema-version]]<<outbox-event-router-property-table-field-event-schema-version, `table.field.event.schema.version`>>
|
|Table, Schema
|When set, it is used as schema version as in the https://kafka.apache.org/20/javadoc/org/apache/kafka/connect/data/ConnectSchema.html#version--[Kafka Connect Schema] javadoc

|[[outbox-event-router-property-route-by-field]]<<outbox-event-router-property-route-by-field, `route.by.field`>>
|`aggregatetype`
|Router
|The column which determines how the events will be routed, the value will become part of the topic name

|[[outbox-event-router-property-route-topic-regex]]<<outbox-event-router-property-route-topic-regex, `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`

|[[outbox-event-router-property-route-topic-replacement]]<<outbox-event-router-property-route-topic-replacement, `route.topic.replacement`>>
|`outbox.event{zwsp}.pass:[${routedByValue}]`
|Router
|The name of the topic in which the events will be routed, a replacement `pass:[${routedByValue}]` is available which is the value of The column configured via `route.by.field`

|[[outbox-event-router-property-route-tombstone-on-empty-payload]]<<outbox-event-router-property-route-tombstone-on-empty-payload, `route.tombstone.on.empty.payload`>>
|`false`
|Router
|Whether or not an empty or `null` payload should cause a tombstone event.

|[[outbox-event-router-property-debezium-op-invalid-behavior]]<<outbox-event-router-property-debezium-op-invalid-behavior, `debezium.op.invalid.behavior`>>
|`warn`
|{prodname}
|While {prodname} is monitoring the table, it is 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`
|===

=== Default table columns

[source]
----
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                  |
----

=== Default configuration values explained

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

[cols="30%a,70%a"]
|===
|Table Column
|Effect

|`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 <<outbox-event-router-property-route-topic-replacement, `route.topic.replacement`>>)

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

|`type`
|The `type` column becomes either part of the message envelope as `eventType` or message header. +
See option `table.fields.additional.placement` for more details.

|`payload`
|The JSON representation of the event itself, becomes either part of the message as `payload` or if other metadata including `eventType` are delivered as headers then the payload becomes the message itself without an encapsulation in an envelope
|===


=== Basic configuration

[source]
----
transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
----

=== Using Avro as the payload format

The outbox routing SMT supports arbitrary payload formats, as the payload column value is passed on transparently.
As an alternative to working with JSON as shown above it is therefore also possible to use Avro.
This can be beneficial for the purposes of message format governance and making sure outbox event schemas evolve in a backwards-compatible way.

How a source application produces Avro messages as an outbox event payload is out of the scope of this documentation.
One possibility could be to leverage the `KafkaAvroSerializer` class and use it to serialize `GenericRecord` instances.
In order to ensure that the Kafka message value is the exact Avro binary data,
apply the following configuration to the connector:

[source]
----
transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.table.fields.additional.placement=type:header:eventType
value.converter=io.debezium.converters.ByteBufferConverter
----

This moves the `eventType` value into a Kafka message header, leaving only the `payload` column value (the Avro data) as the sole message value.
Using `ByteBufferConverter` as the value converter will propagate that value as-is into the Kafka message value.
