Repository navigation
[Feature Request]: Enable Transaction Metadata Events in DebeziumIO JSON Output #40784
Description
What would you like to happen?
Debezium can generates event that represent transaction boundaries and that can enrich data change event message. Find the details here: https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-transaction-metadata
Below are the steps to see the transaction metadata events:
Step 1. Set provide.transaction.metadata = true in the connector.
Step 2. Run the following SQL query:
BEGIN;
INSERT INTO inventory.customers (id, first_name, last_name, email) VALUES (5001, 'Ann', 'One', 'ann.one@example.com');
INSERT INTO inventory.customers (id, first_name, last_name, email) VALUES (5002, 'Bob', 'Two', 'bob.two@example.com');
UPDATE inventory.customers SET last_name = 'Uno' WHERE id = 5001;
COMMIT;Step 3. Monitor the connector output, and you will observe the events listed below:
{"schema":{"type":"struct","fields":[{"type":"string","optional":false,"field":"status"},{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":true,"field":"event_count"},{"type":"array","items":{"type":"struct","fields":[{"type":"string","optional":false,"field":"data_collection"},{"type":"int64","optional":false,"field":"event_count"}],"optional":true,"name":"event.collection","version":1},"optional":true,"field":"data_collections"},{"type":"int64","optional":false,"field":"ts_ms"}],"optional":false,"name":"io.debezium.connector.common.TransactionMetadataValue","version":1},"payload":{"status":"BEGIN","id":"796:48605280","event_count":null,"data_collections":null,"ts_ms":1792911556020}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"tutorial.inventory.customers.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"tutorial.inventory.customers.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,first,first_in_data_collection,last_in_data_collection,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"},{"type":"string","optional":true,"field":"origin"},{"type":"int64","optional":true,"field":"origin_lsn"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","version":1,"field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"tutorial.inventory.customers.Envelope","version":2},"payload":{"before":null,"after":{"id":5001,"first_name":"Ann","last_name":"One","email":"ann.one@example.com"},"source":{"version":"3.5.2.Final","connector":"postgresql","name":"tutorial","ts_ms":1792911556020,"snapshot":"false","db":"inventory","sequence":"[\"48270936\",\"48605280\"]","ts_us":1792911556020003,"ts_ns":1792911556020003000,"schema":"inventory","table":"customers","txId":796,"lsn":48605280,"xmin":null,"origin":null,"origin_lsn":null},"transaction":{"id":"796:48605280","total_order":1,"data_collection_order":1},"op":"c","ts_ms":1792911556229,"ts_us":1792911556229128,"ts_ns":1792911556229128294}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"tutorial.inventory.customers.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"tutorial.inventory.customers.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,first,first_in_data_collection,last_in_data_collection,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"},{"type":"string","optional":true,"field":"origin"},{"type":"int64","optional":true,"field":"origin_lsn"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","version":1,"field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"tutorial.inventory.customers.Envelope","version":2},"payload":{"before":null,"after":{"id":5002,"first_name":"Bob","last_name":"Two","email":"bob.two@example.com"},"source":{"version":"3.5.2.Final","connector":"postgresql","name":"tutorial","ts_ms":1792911556020,"snapshot":"false","db":"inventory","sequence":"[\"48270936\",\"48605840\"]","ts_us":1792911556020003,"ts_ns":1792911556020003000,"schema":"inventory","table":"customers","txId":796,"lsn":48605840,"xmin":null,"origin":null,"origin_lsn":null},"transaction":{"id":"796:48605840","total_order":2,"data_collection_order":2},"op":"c","ts_ms":1792911556230,"ts_us":1792911556230205,"ts_ns":1792911556230205544}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"tutorial.inventory.customers.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"default":0,"field":"id"},{"type":"string","optional":false,"field":"first_name"},{"type":"string","optional":false,"field":"last_name"},{"type":"string","optional":false,"field":"email"}],"optional":true,"name":"tutorial.inventory.customers.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,first,first_in_data_collection,last_in_data_collection,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"},{"type":"string","optional":true,"field":"origin"},{"type":"int64","optional":true,"field":"origin_lsn"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","version":1,"field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"tutorial.inventory.customers.Envelope","version":2},"payload":{"before":{"id":5001,"first_name":"Ann","last_name":"One","email":"ann.one@example.com"},"after":{"id":5001,"first_name":"Ann","last_name":"Uno","email":"ann.one@example.com"},"source":{"version":"3.5.2.Final","connector":"postgresql","name":"tutorial","ts_ms":1792911556020,"snapshot":"false","db":"inventory","sequence":"[\"48270936\",\"48606072\"]","ts_us":1792911556020003,"ts_ns":1792911556020003000,"schema":"inventory","table":"customers","txId":796,"lsn":48606072,"xmin":null,"origin":null,"origin_lsn":null},"transaction":{"id":"796:48606072","total_order":3,"data_collection_order":3},"op":"u","ts_ms":1792911556230,"ts_us":1792911556230738,"ts_ns":1792911556230738627}}
{"schema":{"type":"struct","fields":[{"type":"string","optional":false,"field":"status"},{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":true,"field":"event_count"},{"type":"array","items":{"type":"struct","fields":[{"type":"string","optional":false,"field":"data_collection"},{"type":"int64","optional":false,"field":"event_count"}],"optional":true,"name":"event.collection","version":1},"optional":true,"field":"data_collections"},{"type":"int64","optional":false,"field":"ts_ms"}],"optional":false,"name":"io.debezium.connector.common.TransactionMetadataValue","version":1},"payload":{"status":"END","id":"796:48606256","event_count":3,"data_collections":[{"data_collection":"inventory.customers","event_count":3}],"ts_ms":1792911556020}}Reason
Transaction metadata is useful, particularly when we need to process events while honoring their sequence. In distributed processing, there is no assurance that events will be handled in order based on database commit time. Consequently, I require this feature to be enabled in DebeziumIO.
Proposal
Below are sample transaction metadata events handled by these changes:
Transaction boundary events:
{
"status" : "BEGIN",
"id" : "781:39950456",
"ts_ms" : 1792926401751,
"event_count" : null,
"data_collections" : null
}Payload with Transaction metadata:
{
"metadata" : {
"connector" : "postgresql",
"version" : "3.5.2.Final",
"name" : "beam-debezium-connector",
"database" : "inventory",
"schema" : "inventory",
"table" : "customers",
"op" : "c",
"ts_ms" : 1792926173482,
"snapshot" : "false",
"sequence" : "[null,\"39950456\"]",
"ts_us" : 1792926173482147,
"ts_ns" : 1792926173482147000,
"txId" : 781,
"lsn" : 39950456,
"xmin" : null,
"origin" : null,
"origin_lsn" : null
},
"before" : null,
"after" : {
"fields" : {
"last_name" : "One",
"id" : 5001,
"first_name" : "Ann",
"email" : "ann.one@example.com"
}
},
"transaction" : {
"id" : "781:39950456",
"total_order" : 1,
"data_collection_order" : 1
}
}Issue Priority
Priority: 2 (default / most feature requests should be filed as P2)
Issue Components
- Component: Python SDK
- Component: Java SDK
- Component: Go SDK
- Component: Typescript SDK
- Component: IO connector
- Component: Beam YAML
- Component: Beam examples
- Component: Beam playground
- Component: Beam katas
- Component: Website
- Component: Infrastructure
- Component: Spark Runner
- Component: Flink Runner
- Component: Prism Runner
- Component: Twister2 Runner
- Component: Hazelcast Jet Runner
- Component: Google Cloud Dataflow Runner
Activity