[ilyasahsan123]

Addresses #40784

[github-actions]

📋 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/END transaction boundary event handling (isTransactionMetadataEvent and loadTransactionMetadata) is currently only exercised by the Testcontainers integration test; adding a unit test in SourceRecordJsonTest would 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.

[github-actions]

🟡 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.

[tkaymak]

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:

1. SourceRecordJson.java:153-158 and :224-235: what is the consideration of gating the transaction block behind fullMetadata while BEGIN/END events are emitted regardless? The block only exists when the user opted in via provide.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.

2. SourceRecordJson.java:249-269: consider iterating value.schema().fields() as loadBefore and loadAfter do instead of hard coding the five names. transaction.metadata.factory lets users swap the struct maker, and a missing field would throw DataException here. Same for the transaction block in loadMetadataExtras: transaction.get(field) is not passed through toJsonFriendly, so a nested Struct would be reflected by Gson.

3. 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.

4. 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 the transaction key next to the withFullMetadata paragraph.

5. 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.md line until it is approved (so it's not committed here yet), and please mention that it is also available from Python and Go via connection_properties.

6. Python and Go reach this through the expansion service (DebeziumTransformRegistrar builds readAsJson() and forwards connection_properties), so they get the feature with provide.transaction.metadata=true in connection_properties. The full_metadata docs in sdks/python/apache_beam/io/debezium.py and sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go list the metadata fields but not the transaction key or the BEGIN/END payloads. A one line update in both would help, here or as a follow up.

[ilyasahsan123]

Thank you for your feedbacks, @tkaymak 🙌
I'll address it and notify you when it's ready 👍