[ilyasahsan123]

Addresses #40784

Supports Debezium's transaction metadata (provide.transaction.metadata=true) in DebeziumIO.

Today, the transaction BEGIN/END events have no source block, so SourceRecordJson throws an IllegalArgumentException and the pipeline fails.

With this change:

• Transaction events are mapped to their own payload:

{"status":"BEGIN","id":"781:39950456","ts_ms":1792926401751,"event_count":null,"data_collections":null}                                                                                                                              
{"status":"END","id":"781:39950456","ts_ms":1792926401760,"event_count":3,"data_collections":[{"data_collection":"inventory.customers","event_count":3}]}

• Change events have a new transaction key:

The change event now includes a "transaction" key when there's one.

{"metadata":{...},"before":{...},"after":{"fields":{...}}}
{"metadata":{...},"before":{...},"after":{"fields":{...}},"transaction":{"id":"781:39950456","total_order":1,"data_collection_order":1}}

Python and Go can use this with no code changes: pass provide.transaction.metadata=true via connection_properties (Python) or debeziumio.ConnectionProperties (Go).

CHANGES.md:
I'll update this line https://github.com/apache/beam/blob/master/CHANGES.md?plain=1#L70 with the following content:

(Java/Python/Go) DebeziumIO: readAsJson() can now include op and the remaining Debezium source fields (e.g. MySQL file, pos, row, ts_ms; PostgreSQL lsn, txId, ts_ms) in metadata via withFullMetadata(true) / full_metadata=True / debeziumio.FullMetadata(). Off by default (#40701). readAsJson() also supports transaction metadata events via provide.transaction.metadata=true (Python/Go: connection_properties) (#40784).

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• If this contribution is large, please file an Apache Individual Contributor License Agreement.
• Replace these instructions with a description of your change and why you have made it.
• Mention the appropriate issue in your description (for example: addresses #123 or fixes #456).
• Update CHANGES.md with noteworthy changes.

See CONTRIBUTING.md for more details.

[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 👍

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 56.39%. Comparing base (0f1ac43) to head (92d3e10).
⚠️ Report is 15 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff             @@
##           master   #40785       +/-   ##
===========================================
+ Coverage        0   56.39%   +56.39%     
===========================================
  Files           0      876      +876     
  Lines           0   154789   +154789     
===========================================
+ Hits            0    87296    +87296     
- Misses          0    65636    +65636     
- Partials        0     1857     +1857     

Flag	Coverage Δ
go	28.91% <ø> (?)
python	79.81% <ø> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[ilyasahsan123]

Thanks for catching this. You're right, the transaction block should not depend on fullMetadata. I've moved it so the output now matches the Debezium payload. The block is only filled in when the user sets provide.transaction.metadata=true, so the default output does not change.

To support this, I added two new types:

• Transaction: added to each change event, so the event can be linked to its transaction.
• TransactionEvent: the BEGIN and END events, emitted only when provide.transaction.metadata=true.

With provide.transaction.metadata=true

BEGIN event:

{
  "status" : "BEGIN",
  "id" : "781:39950456",
  "ts_ms" : 1793024470210,
  "event_count" : null,
  "data_collections" : null
}

END event:

{
  "status" : "END",
  "id" : "781:39951784",
  "ts_ms" : 1793024470210,
  "event_count" : 3,
  "data_collections" : [ {
    "data_collection" : "inventory.customers",
    "event_count" : 3
  } ]
}

Change event (note the transaction block at the end):

{
  "metadata" : {
    "connector" : "postgresql",
    "version" : "3.5.2.Final",
    "name" : "beam-debezium-connector",
    "database" : "inventory",
    "schema" : "inventory",
    "table" : "customers",
    "op" : "c",
    "ts_ms" : 1793024873511,
    "snapshot" : "false",
    "sequence" : "[null,\"39950456\"]",
    "ts_us" : 1793024873511455,
    "ts_ns" : 1793024873511455000,
    "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
  }
}

With provide.transaction.metadata=false (default)

No BEGIN/END events are emitted, and transaction is null:

{
  "metadata" : {
    "connector" : "postgresql",
    "version" : "3.5.2.Final",
    "name" : "beam-debezium-connector",
    "database" : "inventory",
    "schema" : "inventory",
    "table" : "customers",
    "op" : "c",
    "ts_ms" : 1793025206974,
    "snapshot" : "false",
    "sequence" : "[null,\"39950456\"]",
    "ts_us" : 1793025206974908,
    "ts_ns" : 1793025206974908000,
    "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" : null
}

[ilyasahsan123]

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.

Thanks for the review. The flat map approach may not provide users with a stable, documented schema.

Please tell me if you would prefer an iterative approach; I can shift it from a field to a flat map.

[ilyasahsan123]

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

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.

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.

I have addressed the concerns noted above. When you have a moment, please review this pull request again. Thank you.

[tkaymak]

Thanks for the quick turnaround, the rework reads well and the BEGIN and END unit test is exactly what was missing.

On typed versus iterating, keep the typed Transaction and TransactionEvent classes. A stable documented schema is the better contract and the names match Debezium. Only read the fields defensively, for example transaction.schema().field("id") != null ? transaction.getString("id") : null, so a custom transaction.metadata.factory yields null instead of a DataException. Since the values are read as primitives the Gson reflection concern is gone.

One thing has to change before I can approve. With serializeNulls() every change event now ends with "transaction":null, which is a behavior change of the default output. Beam's own tests show it, the PostCommit Python runs (3.11, 3.13, 3.14) fail in test_xlang_debezium_read with an unexpected 'transaction': None, and XVR Direct 3.14 fails TestDebeziumIO_BasicRead on the exact string. User pipelines that compare against the old JSON or parse it with a strict schema would break the same way. Please omit the key unless the record has a transaction block. The smallest fix is in Event.toJson(Map): always build the tree with toJsonTree and remove("transaction") when it is null. Then the three golden strings in SourceRecordJsonTest, the MySQL IT and the Postgres IT go back to their previous values, the README sentence becomes "the key is absent" instead of null, and the Python and Go docstrings are accurate as written.

[tkaymak]

Everything else is addressed, thanks for the slot note and the README, Python and Go updates. The Debezium PreCommit ran the six ITs including the new one. Once the null key is gone and the two PostCommits are green I will approve, CHANGES.md stays held back until then. @ilyasahsan123

[github-actions]

Assigning reviewers:

R: @lostluck for label go.
Backup reviewer: shunping (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[ilyasahsan123]

Thanks for the review, @tkaymak! I've addressed everything in the 7754fd0e945. Below are the details:

• The Transaction and TransactionMetadata fields are now read defensively. I added two test that cover missing fields.

• Change events no longer include the Transaction key when there is no transaction block. The README now says the key is absent rather than null.

[tkaymak]

Thanks, the null key is gone, the fields are read defensively and the two new tests cover the missing field cases. PostCommit Python 3.11 to 3.14, XVR Direct and the Debezium PreCommit are green. LGTM.

Please add a CHANGES.md entry for the transaction metadata support under New Features / Improvements, as mentioned earlier.

[ilyasahsan123]

Thank you, @tkaymak!

Also, I have added a CHANGES.md for transaction metadata support under 2.78.0 I/Os and New Features / Improvements as you mentioned. Let me know if there's anything I can do to help.

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.