Repository navigation
[DebeziumIO] Enable Transaction Metadata Events in DebeziumIO JSON Output - #40785
ilyasahsan123 wants to merge 1 commit into
Conversation
Signed-off-by: Ilyas Ahsan <meilyasahsan@gmail.com>
There was a problem hiding this comment.
📋 Review Summary
This PR adds support for Debezium transaction metadata in SourceRecordJson, including mapping BEGIN/END transaction boundary events and including the transaction block in change event metadata when fullMetadata is enabled. The implementation and PostgreSQL integration test are clean and well-structured.
🔍 General Feedback
- The new
BEGIN/ENDtransaction boundary event handling (isTransactionMetadataEventandloadTransactionMetadata) is currently only exercised by the Testcontainers integration test; adding a unit test inSourceRecordJsonTestwould ensure this code path is covered during standard unit test runs.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
| + "\"transaction\":{\"id\":\"796:48630784\",\"total_order\":1," | ||
| + "\"data_collection_order\":1}}," | ||
| + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}}}", | ||
| new SourceRecordJson(transactionRecord, true).toJson()); |
There was a problem hiding this comment.
🟡 Medium: testFullMetadata verifies the transaction block on data change events, but there is no unit test in SourceRecordJsonTest covering transaction BEGIN and END events (isTransactionMetadataEvent and loadTransactionMetadata()). Consider adding a unit test with io.debezium.connector.common.TransactionMetadataKey records (both BEGIN with null event_count/data_collections and END with populated data_collections) so this code path is tested without requiring the Docker-based integration test.
There was a problem hiding this comment.
Thanks, this looks good overall, I think! Opt in through provide.transaction.metadata, the default output is unchanged, and the names match Debezium 3.5.2. A few points:
-
SourceRecordJson.java:153-158and:224-235: what is the consideration of gating thetransactionblock behind fullMetadata while BEGIN/END events are emitted regardless? The block only exists when the user opted in viaprovide.transaction.metadata, so including it whenever the envelope has it does not change the default output. Otherwise the boundary events cannot be correlated with the change events without fullMetadata. -
SourceRecordJson.java:249-269: consider iteratingvalue.schema().fields()asloadBeforeandloadAfterdo instead of hard coding the five names.transaction.metadata.factorylets users swap the struct maker, and a missing field would throwDataExceptionhere. Same for thetransactionblock inloadMetadataExtras:transaction.get(field)is not passed throughtoJsonFriendly, so a nested Struct would be reflected by Gson. -
DebeziumIOPostgresSqlConnectorIT.java:216-218: please add a one line note why the replication slot is created before the inserts, the other tests let Debezium create it. -
README.md: Currently the diff is only a trailing newline (which makes sense as it's a draft PR). You could document the BEGIN/END payload and thetransactionkey next to thewithFullMetadataparagraph. -
Please fill in the PR description and prepare a CHANGES.md line under 2.78.0 I/Os, feel free to extend the existing DebeziumIO entry from #40743. Hold the
CHANGES.mdline until it is approved (so it's not committed here yet), and please mention that it is also available from Python and Go viaconnection_properties. -
Python and Go reach this through the expansion service (
DebeziumTransformRegistrarbuildsreadAsJson()and forwardsconnection_properties), so they get the feature withprovide.transaction.metadata=trueinconnection_properties. Thefull_metadatadocs insdks/python/apache_beam/io/debezium.pyandsdks/go/pkg/beam/io/xlang/debeziumio/debezium.golist the metadata fields but not thetransactionkey or the BEGIN/END payloads. A one line update in both would help, here or as a follow up.
|
Thank you for your feedbacks, @tkaymak 🙌 |
Addresses #40784