From e13f7141b8245d1fa5fc458bcf8ce6efc9fcf6d0 Mon Sep 17 00:00:00 2001 From: Ilyas Ahsan Date: Sun, 25 Oct 2026 13:41:38 +0700 Subject: [PATCH 1/6] [DebeziumIO] Support Transaction Metadata Events Signed-off-by: Ilyas Ahsan --- sdks/java/io/debezium/src/README.md | 2 +- .../apache/beam/io/debezium/DebeziumIO.java | 3 +- .../beam/io/debezium/SourceRecordJson.java | 89 +++++++++++++++++-- .../DebeziumIOPostgresSqlConnectorIT.java | 83 +++++++++++++++++ .../io/debezium/SourceRecordJsonTest.java | 43 +++++++++ 5 files changed, 211 insertions(+), 9 deletions(-) diff --git a/sdks/java/io/debezium/src/README.md b/sdks/java/io/debezium/src/README.md index cd5e4ceabacb..9f76cfdf7826 100644 --- a/sdks/java/io/debezium/src/README.md +++ b/sdks/java/io/debezium/src/README.md @@ -172,4 +172,4 @@ You can run Integration Tests using **gradlew**. Example of running the MySQL Connector Integration Test: ``` ./gradlew integrationTest -p sdks/java/io/debezium/ --tests org.apache.beam.io.debezium.DebeziumIOMySqlConnectorIT -DintegrationTestRunner=direct -``` \ No newline at end of file +``` diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java index cc31c451e9c8..22f4b5e317ef 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java @@ -312,7 +312,8 @@ public Read withOffsetRetainer(OffsetRetainer retainer) { /** * Adds {@code op} and the remaining Debezium {@code source} fields (e.g. MySQL {@code file}, * {@code pos}, {@code row}, {@code ts_ms}; PostgreSQL {@code lsn}, {@code txId}, {@code ts_ms}) - * to {@code metadata} when reading as JSON. + * to {@code metadata} when reading as JSON, plus the change event's {@code transaction} block + * as a nested object when the connector has {@code provide.transaction.metadata=true}. * *

Only applies to {@link DebeziumIO#readAsJson()} / {@link * SourceRecordJson.SourceRecordJsonMapper}; combining it with a custom {@link diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java index 694b232d4647..d0e70f9d7b1c 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java @@ -19,9 +19,11 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.Serializable; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; import java.util.Set; import org.apache.beam.vendor.grpc.v1p69p0.com.google.gson.Gson; @@ -29,6 +31,7 @@ import org.apache.beam.vendor.grpc.v1p69p0.com.google.gson.JsonObject; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet; import org.apache.kafka.connect.data.Field; +import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.errors.DataException; import org.apache.kafka.connect.source.SourceRecord; @@ -72,10 +75,20 @@ *

  • Every other field of the Debezium {@code source} block, in schema order, that is not * already represented above (for example {@code ts_ms}, {@code snapshot}, MySQL {@code file}, * {@code pos}, {@code row}, or PostgreSQL {@code lsn}, {@code txId}) + *
  • transaction - The Debezium {@code transaction} block of the change event ({@code id}, + * {@code total_order} and {@code data_collection_order}) as a nested object, if the connector + * has {@code provide.transaction.metadata=true} * * *

    By default ({@code fullMetadata = false}) the output only contains the six keys above. * + *

    Transaction metadata

    + * + *

    When the connector has {@code provide.transaction.metadata=true}, Debezium also emits a {@code + * BEGIN} and an {@code END} event around the change events of each transaction. These have no + * {@code source}, {@code before} or {@code after}, so they are mapped to their own payload instead: + * {@code status}, {@code id}, {@code ts_ms}, {@code event_count} and {@code data_collections}. + * *

    Usage Example

    * *

    Map each SourceRecord to JSON @@ -98,11 +111,18 @@ public class SourceRecordJson { * existing metadata key), so they are never added as full metadata extras. */ private static final Set RESERVED_METADATA_KEYS = - ImmutableSet.of("connector", "version", "name", "db", "database", "schema", "table", "op"); + ImmutableSet.of( + "connector", "version", "name", "db", "database", "schema", "table", "op", "transaction"); + + private static final String SCHEMA_TRANSACTION_METADATA_EVENT_KEY_NAME = + "io.debezium.connector.common.TransactionMetadataKey"; + + private static final Gson GSON = new GsonBuilder().serializeNulls().create(); private final @Nullable Struct value; private final @Nullable Event event; private final Map metadataExtras; + private final @Nullable Map transactionMetadata; /** * Initializer. @@ -130,7 +150,13 @@ public SourceRecordJson(@Nullable SourceRecord sourceRecord, boolean fullMetadat if (this.value == null) { this.event = new Event(null, null, null); this.metadataExtras = Collections.emptyMap(); + this.transactionMetadata = null; + } else if (isTransactionMetadataEvent(sourceRecord.keySchema())) { + this.event = null; + this.metadataExtras = Collections.emptyMap(); + this.transactionMetadata = this.loadTransactionMetadata(); } else { + this.transactionMetadata = null; Event.Metadata metadata = this.loadMetadata(); Event.Before before = this.loadBefore(); Event.After after = this.loadAfter(); @@ -179,7 +205,7 @@ private Event.Metadata loadMetadata() { /** * Extracts the full metadata extras: the envelope's {@code op} (if the value schema has one), * then every field of the Debezium {@code source} block, in schema order, that isn't already - * represented in {@link Event.Metadata}. + * represented in {@link Event.Metadata}, then the envelope's {@code transaction} block (if set). * * @return extra metadata keys, in insertion order */ @@ -195,6 +221,18 @@ private Map loadMetadataExtras() { extras.put(fieldName, toJsonFriendly(source.get(field))); } } + + Struct transaction = + this.value.schema().field("transaction") != null + ? this.value.getStruct("transaction") + : null; + if (transaction != null) { + Map transactionFields = new LinkedHashMap<>(); + for (Field field : transaction.schema().fields()) { + transactionFields.put(field.name(), transaction.get(field)); + } + extras.put("transaction", transactionFields); + } return extras; } @@ -203,6 +241,33 @@ private Map loadMetadataExtras() { return value instanceof Struct ? String.valueOf(value) : value; } + /** + * Extracts the payload of a transaction BEGIN/END event. + * + * @return status, id, ts_ms, event_count and data_collections, in insertion order + */ + private Map loadTransactionMetadata() { + List> dataCollections = null; + List dataCollectionStructs = this.value.getArray("data_collections"); + if (dataCollectionStructs != null) { + dataCollections = new ArrayList<>(); + for (Struct dataCollectionStruct : dataCollectionStructs) { + Map dataCollection = new LinkedHashMap<>(); + dataCollection.put("data_collection", dataCollectionStruct.getString("data_collection")); + dataCollection.put("event_count", dataCollectionStruct.getInt64("event_count")); + dataCollections.add(dataCollection); + } + } + + Map transactionMetadata = new LinkedHashMap<>(); + transactionMetadata.put("status", this.value.getString("status")); + transactionMetadata.put("id", this.value.getString("id")); + transactionMetadata.put("ts_ms", this.value.getInt64("ts_ms")); + transactionMetadata.put("event_count", this.value.getInt64("event_count")); + transactionMetadata.put("data_collections", dataCollections); + return transactionMetadata; + } + /** * Extracts the before field within SourceRecord. * @@ -257,6 +322,9 @@ private Event.After loadAfter() { * @return JSON String */ public String toJson() { + if (this.transactionMetadata != null) { + return GSON.toJson(this.transactionMetadata); + } return this.event.toJson(this.metadataExtras); } @@ -325,17 +393,16 @@ public String toJson() { * @return JSON String */ String toJson(Map extras) { - Gson gson = new GsonBuilder().serializeNulls().create(); if (extras.isEmpty() || metadata == null) { - return gson.toJson(this); + return GSON.toJson(this); } // Gson can't flatten a map into an object, so append the extras to the metadata tree. - JsonObject tree = gson.toJsonTree(this).getAsJsonObject(); + JsonObject tree = GSON.toJsonTree(this).getAsJsonObject(); JsonObject metadataJson = tree.getAsJsonObject("metadata"); for (Map.Entry extra : extras.entrySet()) { - metadataJson.add(extra.getKey(), gson.toJsonTree(extra.getValue())); + metadataJson.add(extra.getKey(), GSON.toJsonTree(extra.getValue())); } - return gson.toJson(tree); + return GSON.toJson(tree); } /** Depicts the metadata within a SourceRecord. It has valuable fields. */ @@ -402,4 +469,12 @@ public After(@Nullable Map fields) { } } } + + /** + * Whether the record is a transaction BEGIN/END event, emitted when {@code + * provide.transaction.metadata} is enabled. These are identified by their key schema name. + */ + private static boolean isTransactionMetadataEvent(@Nullable Schema keySchema) { + return keySchema != null && SCHEMA_TRANSACTION_METADATA_EVENT_KEY_NAME.equals(keySchema.name()); + } } diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java index 0a76415d23ba..59c0ea75d322 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java +++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java @@ -18,10 +18,15 @@ package org.apache.beam.io.debezium; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.allOf; +import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.hasItem; import io.debezium.connector.postgresql.PostgresConnector; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; import javax.sql.DataSource; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.coders.StringUtf8Coder; @@ -195,4 +200,82 @@ public void testDebeziumIOPostgresSql() { p.run().waitUntilFinish(); POSTGRES_SQL_CONTAINER.stop(); } + + /** + * Debezium - PostgresSql connector Test with transaction metadata. + * + *

    Tests that the transaction BEGIN/END events emitted with {@code + * provide.transaction.metadata=true}, which have no {@code source} block, are mapped to JSON + * alongside the data change events of the transaction, and that with full metadata each change + * event carries its {@code transaction} block. + */ + @Test + public void testShouldProcessTransactionMetadata() throws SQLException { + POSTGRES_SQL_CONTAINER.start(); + + try (Connection connection = getPostgresDatasource().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute( + "SELECT pg_create_logical_replication_slot('beam_transaction_metadata', 'decoderbufs')"); + connection.setAutoCommit(false); + statement.executeUpdate( + "INSERT INTO inventory.customers (id, first_name, last_name, email) " + + "VALUES (5001, 'Ann', 'One', 'ann.one@example.com')"); + statement.executeUpdate( + "INSERT INTO inventory.customers (id, first_name, last_name, email) " + + "VALUES (5002, 'Bob', 'Two', 'bob.two@example.com')"); + statement.executeUpdate("UPDATE inventory.customers SET last_name = 'Uno' WHERE id = 5001"); + connection.commit(); + } + + String host = POSTGRES_SQL_CONTAINER.getHost(); + String port = POSTGRES_SQL_CONTAINER.getMappedPort(5432).toString(); + + PipelineOptions options = PipelineOptionsFactory.create(); + Pipeline p = Pipeline.create(options); + PCollection results = + p.apply( + DebeziumIO.readAsJson() + .withConnectorConfiguration( + DebeziumIO.ConnectorConfiguration.create() + .withUsername("debezium") + .withPassword("dbz") + .withConnectorClass(PostgresConnector.class) + .withHostName(host) + .withPort(port) + .withConnectionProperty("database.dbname", "inventory") + .withConnectionProperty("table.include.list", "inventory.customers") + .withConnectionProperty("snapshot.mode", "no_data") + .withConnectionProperty("provide.transaction.metadata", "true") + .withConnectionProperty("slot.name", "beam_transaction_metadata")) + .withFullMetadata(true) + .withMaxNumberOfRecords(5) + .withMaxTimeToRun(120_000L)); + + PAssert.that(results) + .satisfies( + (Iterable res) -> { + assertThat(Lists.newArrayList(res).size(), equalTo(5)); + assertThat(res, hasItem(containsString("{\"status\":\"BEGIN\","))); + assertThat(res, hasItem(containsString("\"email\":\"ann.one@example.com\""))); + assertThat(res, hasItem(containsString("\"email\":\"bob.two@example.com\""))); + assertThat( + res, + hasItem( + allOf( + containsString("\"last_name\":\"Uno\""), + containsString("\"transaction\":{\"id\":\""), + containsString("\"total_order\":3,\"data_collection_order\":3}")))); + assertThat( + res, + hasItem( + containsString( + "\"event_count\":3,\"data_collections\":" + + "[{\"data_collection\":\"inventory.customers\",\"event_count\":3}]"))); + return null; + }); + + p.run().waitUntilFinish(); + POSTGRES_SQL_CONTAINER.stop(); + } } diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java index bcb33e97e4df..575c519555b6 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java +++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java @@ -76,6 +76,39 @@ public void testFullMetadata() { .put("file", "mysql-bin.000003") .put("pos", 796L) .put("row", 0); + + Schema transactionSchema = + SchemaBuilder.struct() + .optional() + .name("event.block") + .field("id", Schema.STRING_SCHEMA) + .field("total_order", Schema.INT64_SCHEMA) + .field("data_collection_order", Schema.INT64_SCHEMA) + .build(); + Struct transaction = + new Struct(transactionSchema) + .put("id", "796:48630784") + .put("total_order", 1L) + .put("data_collection_order", 1L); + Schema rowSchema = SchemaBuilder.struct().optional().field("id", Schema.INT32_SCHEMA).build(); + Struct source = postgresSource(); + Schema envelopeSchema = + SchemaBuilder.struct() + .field("before", rowSchema) + .field("after", rowSchema) + .field("source", source.schema()) + .field("op", Schema.STRING_SCHEMA) + .field("transaction", transactionSchema) + .build(); + Struct envelope = + new Struct(envelopeSchema) + .put("after", new Struct(rowSchema).put("id", 1001)) + .put("source", source) + .put("op", "c") + .put("transaction", transaction); + SourceRecord transactionRecord = new SourceRecord(null, null, null, envelopeSchema, envelope); + + // MySQL has no "schema" source field, so the binlog "file" is also exposed as "schema". assertEquals( "{\"metadata\":{\"connector\":\"mysql\",\"version\":\"3.5.2.Final\",\"name\":\"n\"," + "\"database\":\"inventory\",\"schema\":\"mysql-bin.000003\",\"table\":\"customers\"," @@ -90,6 +123,16 @@ public void testFullMetadata() { + "\"op\":\"u\",\"ts_ms\":1791689147000,\"lsn\":37094192}," + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}}}", new SourceRecordJson(buildEnvelopeRecord(postgresSource(), "u"), true).toJson()); + + // With provide.transaction.metadata, the envelope's transaction block is included. + assertEquals( + "{\"metadata\":{\"connector\":\"postgresql\",\"version\":\"3.5.2.Final\",\"name\":\"n\"," + + "\"database\":\"postgres\",\"schema\":\"inventory\",\"table\":\"customers\"," + + "\"op\":\"c\",\"ts_ms\":1791689147000,\"lsn\":37094192," + + "\"transaction\":{\"id\":\"796:48630784\",\"total_order\":1," + + "\"data_collection_order\":1}}," + + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}}}", + new SourceRecordJson(transactionRecord, true).toJson()); } @Test From 0a5bd0ac770457c99fd8c4664f181e1a71f59f0f Mon Sep 17 00:00:00 2001 From: Ilyas Ahsan Date: Mon, 26 Oct 2026 20:07:14 +0700 Subject: [PATCH 2/6] [DebeziumIO] Rework transaction metadata in SourceRecordJson Signed-off-by: Ilyas Ahsan --- .../apache/beam/io/debezium/DebeziumIO.java | 3 +- .../beam/io/debezium/SourceRecordJson.java | 219 +++++++++++++----- .../debezium/DebeziumIOMySqlConnectorIT.java | 2 +- .../DebeziumIOPostgresSqlConnectorIT.java | 2 +- .../io/debezium/SourceRecordJsonTest.java | 70 +++++- 5 files changed, 232 insertions(+), 64 deletions(-) diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java index 22f4b5e317ef..cc31c451e9c8 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumIO.java @@ -312,8 +312,7 @@ public Read withOffsetRetainer(OffsetRetainer retainer) { /** * Adds {@code op} and the remaining Debezium {@code source} fields (e.g. MySQL {@code file}, * {@code pos}, {@code row}, {@code ts_ms}; PostgreSQL {@code lsn}, {@code txId}, {@code ts_ms}) - * to {@code metadata} when reading as JSON, plus the change event's {@code transaction} block - * as a nested object when the connector has {@code provide.transaction.metadata=true}. + * to {@code metadata} when reading as JSON. * *

    Only applies to {@link DebeziumIO#readAsJson()} / {@link * SourceRecordJson.SourceRecordJsonMapper}; combining it with a custom {@link diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java index d0e70f9d7b1c..090dd97defaa 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/SourceRecordJson.java @@ -29,6 +29,7 @@ import org.apache.beam.vendor.grpc.v1p69p0.com.google.gson.Gson; import org.apache.beam.vendor.grpc.v1p69p0.com.google.gson.GsonBuilder; import org.apache.beam.vendor.grpc.v1p69p0.com.google.gson.JsonObject; +import org.apache.beam.vendor.grpc.v1p69p0.com.google.gson.annotations.SerializedName; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet; import org.apache.kafka.connect.data.Field; import org.apache.kafka.connect.data.Schema; @@ -60,6 +61,9 @@ *

  • schema - Database Schema (for MySQL, the binlog file name) *
  • table - Table name * + *
  • transaction - {@link #loadTransaction}: the Debezium {@code transaction} block of the + * change event ({@code id}, {@code total_order} and {@code data_collection_order}), or {@code + * null} unless the connector has {@code provide.transaction.metadata=true} * * *

    Full metadata

    @@ -75,9 +79,6 @@ *
  • Every other field of the Debezium {@code source} block, in schema order, that is not * already represented above (for example {@code ts_ms}, {@code snapshot}, MySQL {@code file}, * {@code pos}, {@code row}, or PostgreSQL {@code lsn}, {@code txId}) - *
  • transaction - The Debezium {@code transaction} block of the change event ({@code id}, - * {@code total_order} and {@code data_collection_order}) as a nested object, if the connector - * has {@code provide.transaction.metadata=true} * * *

    By default ({@code fullMetadata = false}) the output only contains the six keys above. @@ -86,8 +87,9 @@ * *

    When the connector has {@code provide.transaction.metadata=true}, Debezium also emits a {@code * BEGIN} and an {@code END} event around the change events of each transaction. These have no - * {@code source}, {@code before} or {@code after}, so they are mapped to their own payload instead: - * {@code status}, {@code id}, {@code ts_ms}, {@code event_count} and {@code data_collections}. + * {@code source}, {@code before} or {@code after}, so they are mapped to their own payload instead + * ({@link #loadTransactionEvent}): {@code status}, {@code id}, {@code ts_ms}, {@code event_count} + * and {@code data_collections}. * *

    Usage Example

    * @@ -121,8 +123,8 @@ public class SourceRecordJson { private final @Nullable Struct value; private final @Nullable Event event; + private final @Nullable TransactionEvent transactionEvent; private final Map metadataExtras; - private final @Nullable Map transactionMetadata; /** * Initializer. @@ -148,20 +150,21 @@ public SourceRecordJson(@Nullable SourceRecord sourceRecord, boolean fullMetadat this.value = (Struct) sourceRecord.value(); if (this.value == null) { - this.event = new Event(null, null, null); + this.event = new Event(null, null, null, null); + this.transactionEvent = null; this.metadataExtras = Collections.emptyMap(); - this.transactionMetadata = null; } else if (isTransactionMetadataEvent(sourceRecord.keySchema())) { this.event = null; + this.transactionEvent = this.loadTransactionEvent(); this.metadataExtras = Collections.emptyMap(); - this.transactionMetadata = this.loadTransactionMetadata(); } else { - this.transactionMetadata = null; Event.Metadata metadata = this.loadMetadata(); Event.Before before = this.loadBefore(); Event.After after = this.loadAfter(); + Event.Transaction transaction = this.loadTransaction(); - this.event = new Event(metadata, before, after); + this.event = new Event(metadata, before, after, transaction); + this.transactionEvent = null; this.metadataExtras = fullMetadata && metadata != null ? this.loadMetadataExtras() : Collections.emptyMap(); } @@ -205,7 +208,7 @@ private Event.Metadata loadMetadata() { /** * Extracts the full metadata extras: the envelope's {@code op} (if the value schema has one), * then every field of the Debezium {@code source} block, in schema order, that isn't already - * represented in {@link Event.Metadata}, then the envelope's {@code transaction} block (if set). + * represented in {@link Event.Metadata}. * * @return extra metadata keys, in insertion order */ @@ -221,18 +224,6 @@ private Map loadMetadataExtras() { extras.put(fieldName, toJsonFriendly(source.get(field))); } } - - Struct transaction = - this.value.schema().field("transaction") != null - ? this.value.getStruct("transaction") - : null; - if (transaction != null) { - Map transactionFields = new LinkedHashMap<>(); - for (Field field : transaction.schema().fields()) { - transactionFields.put(field.name(), transaction.get(field)); - } - extras.put("transaction", transactionFields); - } return extras; } @@ -241,33 +232,6 @@ private Map loadMetadataExtras() { return value instanceof Struct ? String.valueOf(value) : value; } - /** - * Extracts the payload of a transaction BEGIN/END event. - * - * @return status, id, ts_ms, event_count and data_collections, in insertion order - */ - private Map loadTransactionMetadata() { - List> dataCollections = null; - List dataCollectionStructs = this.value.getArray("data_collections"); - if (dataCollectionStructs != null) { - dataCollections = new ArrayList<>(); - for (Struct dataCollectionStruct : dataCollectionStructs) { - Map dataCollection = new LinkedHashMap<>(); - dataCollection.put("data_collection", dataCollectionStruct.getString("data_collection")); - dataCollection.put("event_count", dataCollectionStruct.getInt64("event_count")); - dataCollections.add(dataCollection); - } - } - - Map transactionMetadata = new LinkedHashMap<>(); - transactionMetadata.put("status", this.value.getString("status")); - transactionMetadata.put("id", this.value.getString("id")); - transactionMetadata.put("ts_ms", this.value.getInt64("ts_ms")); - transactionMetadata.put("event_count", this.value.getInt64("event_count")); - transactionMetadata.put("data_collections", dataCollections); - return transactionMetadata; - } - /** * Extracts the before field within SourceRecord. * @@ -316,14 +280,63 @@ private Event.After loadAfter() { return new Event.After(fields); } + /** + * Extracts the transaction field within SourceRecord, set when the connector has {@code + * provide.transaction.metadata=true}. + * + * @return Transaction + */ + private Event.Transaction loadTransaction() { + @Nullable Struct transaction; + try { + transaction = (Struct) this.value.get("transaction"); + } catch (DataException e) { + return null; + } + if (transaction == null) { + return null; + } + + return new Event.Transaction( + transaction.getString("id"), + transaction.getInt64("total_order"), + transaction.getInt64("data_collection_order")); + } + + /** + * Extracts the payload of a transaction BEGIN/END event. + * + * @return TransactionEvent + */ + private TransactionEvent loadTransactionEvent() { + @Nullable List dataCollections = null; + @Nullable List dataCollectionStructs = this.value.getArray("data_collections"); + if (dataCollectionStructs != null) { + dataCollections = new ArrayList<>(); + for (Struct dataCollection : dataCollectionStructs) { + dataCollections.add( + new TransactionEvent.DataCollection( + dataCollection.getString("data_collection"), + dataCollection.getInt64("event_count"))); + } + } + + return new TransactionEvent( + this.value.getString("status"), + this.value.getString("id"), + this.value.getInt64("ts_ms"), + this.value.getInt64("event_count"), + dataCollections); + } + /** * Transforms the extracted data to a JSON string. * * @return JSON String */ public String toJson() { - if (this.transactionMetadata != null) { - return GSON.toJson(this.transactionMetadata); + if (this.transactionEvent != null) { + return this.transactionEvent.toJson(); } return this.event.toJson(this.metadataExtras); } @@ -359,6 +372,7 @@ static class Event implements Serializable { private final SourceRecordJson.Event.Metadata metadata; private final SourceRecordJson.Event.Before before; private final SourceRecordJson.Event.After after; + private final SourceRecordJson.Event.Transaction transaction; /** * Event Initializer. @@ -366,14 +380,17 @@ static class Event implements Serializable { * @param metadata Metadata retrieved from SourceRecord * @param before Before data retrieved from SourceRecord * @param after After data retrieved from SourceRecord + * @param transaction Transaction block retrieved from SourceRecord */ public Event( SourceRecordJson.Event.Metadata metadata, SourceRecordJson.Event.Before before, - SourceRecordJson.Event.After after) { + SourceRecordJson.Event.After after, + SourceRecordJson.Event.Transaction transaction) { this.metadata = metadata; this.before = before; this.after = after; + this.transaction = transaction; } /** @@ -468,6 +485,100 @@ public After(@Nullable Map fields) { this.fields = fields; } } + + /** Depicts the transaction field within SourceRecord. */ + @SuppressWarnings("unused") // Unused params are validated as JSON + static class Transaction implements Serializable { + private final @Nullable String id; + + @SerializedName("total_order") + private final @Nullable Long totalOrder; + + @SerializedName("data_collection_order") + private final @Nullable Long dataCollectionOrder; + + /** + * Transaction Initializer. + * + * @param id Transaction id + * @param totalOrder Position of the event among all events of the transaction + * @param dataCollectionOrder Position of the event among the events of its data collection + */ + public Transaction( + @Nullable String id, @Nullable Long totalOrder, @Nullable Long dataCollectionOrder) { + this.id = id; + this.totalOrder = totalOrder; + this.dataCollectionOrder = dataCollectionOrder; + } + } + } + + /** Depicts a transaction BEGIN/END SourceRecord as a TransactionEvent to be mapped as JSON. */ + @SuppressWarnings("unused") // Unused params are validated as JSON + static class TransactionEvent implements Serializable { + private final @Nullable String status; + private final @Nullable String id; + + @SerializedName("ts_ms") + private final @Nullable Long tsMs; + + @SerializedName("event_count") + private final @Nullable Long eventCount; + + @SerializedName("data_collections") + private final @Nullable List dataCollections; + + /** + * TransactionEvent Initializer. + * + * @param status {@code BEGIN} or {@code END} + * @param id Transaction id + * @param tsMs Time the transaction began or ended, in milliseconds + * @param eventCount Number of events in the transaction, only set on {@code END} + * @param dataCollections Number of events per data collection, only set on {@code END} + */ + public TransactionEvent( + @Nullable String status, + @Nullable String id, + @Nullable Long tsMs, + @Nullable Long eventCount, + @Nullable List dataCollections) { + this.status = status; + this.id = id; + this.tsMs = tsMs; + this.eventCount = eventCount; + this.dataCollections = dataCollections; + } + + /** + * Transforms the TransactionEvent to a JSON string. + * + * @return JSON String + */ + public String toJson() { + return GSON.toJson(this); + } + + /** Depicts an entry of the data_collections field within SourceRecord. */ + @SuppressWarnings("unused") // Unused params are validated as JSON + static class DataCollection implements Serializable { + @SerializedName("data_collection") + private final @Nullable String dataCollection; + + @SerializedName("event_count") + private final @Nullable Long eventCount; + + /** + * DataCollection Initializer. + * + * @param dataCollection Data collection (table) name + * @param eventCount Number of events of the transaction in this data collection + */ + public DataCollection(@Nullable String dataCollection, @Nullable Long eventCount) { + this.dataCollection = dataCollection; + this.eventCount = eventCount; + } + } } /** diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java index 29c8e33ac06d..0a3954c46287 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java +++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java @@ -289,7 +289,7 @@ public void testDebeziumIOMySql() { + "\"database\":\"inventory\",\"schema\":\"mysql-bin.000003\",\"table\":\"addresses\"},\"before\":null," + "\"after\":{\"fields\":{\"zip\":\"76036\",\"city\":\"Euless\"," + "\"street\":\"3183 Moore Avenue\",\"id\":10,\"state\":\"Texas\",\"customer_id\":1001," - + "\"type\":\"SHIPPING\"}}}"; + + "\"type\":\"SHIPPING\"}},\"transaction\":null}"; PAssert.that(results) .satisfies( diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java index 59c0ea75d322..9b8ad714ba8d 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java +++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java @@ -188,7 +188,7 @@ public void testDebeziumIOPostgresSql() { "{\"metadata\":{\"connector\":\"postgresql\",\"version\":\"3.5.2.Final\",\"name\":\"beam-debezium-connector\"," + "\"database\":\"inventory\",\"schema\":\"inventory\",\"table\":\"customers\"},\"before\":null," + "\"after\":{\"fields\":{\"last_name\":\"Thomas\",\"id\":1001,\"first_name\":\"Sally\"," - + "\"email\":\"sally.thomas@acme.com\"}}}"; + + "\"email\":\"sally.thomas@acme.com\"}},\"transaction\":null}"; PAssert.that(results) .satisfies( diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java index 575c519555b6..01fc6f5bd1ff 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java +++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java @@ -23,6 +23,7 @@ import static org.junit.Assert.assertTrue; import java.io.Serializable; +import java.util.Collections; import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaBuilder; import org.apache.kafka.connect.data.Struct; @@ -51,7 +52,8 @@ public void testSourceRecordJson() { + "\"temperature\":104.4,\"childrenAndAge\":null,\"age\":16}}," + "\"after\":{\"fields\":{\"country\":null,\"distance\":123.423,\"birthYear\":null," + "\"name\":\"after-name\"," - + "\"temperature\":104.4,\"childrenAndAge\":null,\"age\":16}}}"; + + "\"temperature\":104.4,\"childrenAndAge\":null,\"age\":16}}," + + "\"transaction\":null}"; assertEquals(expectedJson, jsonString); } @@ -113,7 +115,8 @@ public void testFullMetadata() { "{\"metadata\":{\"connector\":\"mysql\",\"version\":\"3.5.2.Final\",\"name\":\"n\"," + "\"database\":\"inventory\",\"schema\":\"mysql-bin.000003\",\"table\":\"customers\"," + "\"op\":\"c\",\"ts_ms\":1791689147000,\"gtid\":null,\"file\":\"mysql-bin.000003\"," - + "\"pos\":796,\"row\":0},\"before\":null,\"after\":{\"fields\":{\"id\":1001}}}", + + "\"pos\":796,\"row\":0},\"before\":null,\"after\":{\"fields\":{\"id\":1001}}," + + "\"transaction\":null}", new SourceRecordJson(buildEnvelopeRecord(mySql, "c"), true).toJson()); // PostgreSQL "schema" is a source field: it stays the existing key and isn't repeated. @@ -121,20 +124,75 @@ public void testFullMetadata() { "{\"metadata\":{\"connector\":\"postgresql\",\"version\":\"3.5.2.Final\",\"name\":\"n\"," + "\"database\":\"postgres\",\"schema\":\"inventory\",\"table\":\"customers\"," + "\"op\":\"u\",\"ts_ms\":1791689147000,\"lsn\":37094192}," - + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}}}", + + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}},\"transaction\":null}", new SourceRecordJson(buildEnvelopeRecord(postgresSource(), "u"), true).toJson()); // With provide.transaction.metadata, the envelope's transaction block is included. assertEquals( "{\"metadata\":{\"connector\":\"postgresql\",\"version\":\"3.5.2.Final\",\"name\":\"n\"," + "\"database\":\"postgres\",\"schema\":\"inventory\",\"table\":\"customers\"," - + "\"op\":\"c\",\"ts_ms\":1791689147000,\"lsn\":37094192," + + "\"op\":\"c\",\"ts_ms\":1791689147000,\"lsn\":37094192}," + + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}}," + "\"transaction\":{\"id\":\"796:48630784\",\"total_order\":1," - + "\"data_collection_order\":1}}," - + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}}}", + + "\"data_collection_order\":1}}", new SourceRecordJson(transactionRecord, true).toJson()); } + @Test + public void testTransactionMetadataEvent() { + Schema keySchema = + SchemaBuilder.struct() + .name("io.debezium.connector.common.TransactionMetadataKey") + .field("id", Schema.STRING_SCHEMA) + .build(); + Schema dataCollectionSchema = + SchemaBuilder.struct() + .field("data_collection", Schema.STRING_SCHEMA) + .field("event_count", Schema.INT64_SCHEMA) + .build(); + Schema valueSchema = + SchemaBuilder.struct() + .name("io.debezium.connector.common.TransactionMetadataValue") + .field("status", Schema.STRING_SCHEMA) + .field("id", Schema.STRING_SCHEMA) + .field("event_count", Schema.OPTIONAL_INT64_SCHEMA) + .field("data_collections", SchemaBuilder.array(dataCollectionSchema).optional().build()) + .field("ts_ms", Schema.INT64_SCHEMA) + .build(); + Struct key = new Struct(keySchema).put("id", "781:39950456"); + + Struct begin = + new Struct(valueSchema) + .put("status", "BEGIN") + .put("id", "781:39950456") + .put("ts_ms", 1792926401751L); + assertEquals( + "{\"status\":\"BEGIN\",\"id\":\"781:39950456\",\"ts_ms\":1792926401751," + + "\"event_count\":null,\"data_collections\":null}", + new SourceRecordJson(new SourceRecord(null, null, null, keySchema, key, valueSchema, begin)) + .toJson()); + + Struct end = + new Struct(valueSchema) + .put("status", "END") + .put("id", "781:39950456") + .put("event_count", 3L) + .put( + "data_collections", + Collections.singletonList( + new Struct(dataCollectionSchema) + .put("data_collection", "inventory.customers") + .put("event_count", 3L))) + .put("ts_ms", 1792926401760L); + assertEquals( + "{\"status\":\"END\",\"id\":\"781:39950456\",\"ts_ms\":1792926401760," + + "\"event_count\":3,\"data_collections\":" + + "[{\"data_collection\":\"inventory.customers\",\"event_count\":3}]}", + new SourceRecordJson( + new SourceRecord(null, null, null, keySchema, key, valueSchema, end), true) + .toJson()); + } + @Test public void testFullMetadataWithoutOpField() { String json = new SourceRecordJson(buildEnvelopeRecord(postgresSource(), null), true).toJson(); From 9533d793e3bbf0dd50a292021ee76529a7871891 Mon Sep 17 00:00:00 2001 From: Ilyas Ahsan Date: Mon, 26 Oct 2026 20:24:35 +0700 Subject: [PATCH 3/6] [DebeziumIO] Add a one-line note explaining purpose of slot creation Signed-off-by: Ilyas Ahsan --- .../beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java | 1 + 1 file changed, 1 insertion(+) diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java index 9b8ad714ba8d..a3da4d2d497b 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java +++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java @@ -215,6 +215,7 @@ public void testShouldProcessTransactionMetadata() throws SQLException { try (Connection connection = getPostgresDatasource().getConnection(); Statement statement = connection.createStatement()) { + // Create the slot first so it captures the commit below (no_data skips the snapshot). statement.execute( "SELECT pg_create_logical_replication_slot('beam_transaction_metadata', 'decoderbufs')"); connection.setAutoCommit(false); From 69c7347c65d44ee077c9a63a89ab5e6f71982bcc Mon Sep 17 00:00:00 2001 From: Ilyas Ahsan Date: Mon, 26 Oct 2026 20:29:45 +0700 Subject: [PATCH 4/6] [DebeziumIO] Add the transaction events information in README.md Signed-off-by: Ilyas Ahsan --- sdks/java/io/debezium/src/README.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sdks/java/io/debezium/src/README.md b/sdks/java/io/debezium/src/README.md index 9f76cfdf7826..7a3574503b07 100644 --- a/sdks/java/io/debezium/src/README.md +++ b/sdks/java/io/debezium/src/README.md @@ -95,6 +95,8 @@ If you want to use this built-in JSON Mapper, you can do it by setting an instan By default, the JSON `metadata` object holds `connector`, `version`, `name`, `database`, `schema` and `table`. Call `withFullMetadata(true)` on `readAsJson()` to also add the change event's `op` (`c`, `u`, `d`, `r` or `t`) and every other field of the Debezium `source` block to `metadata` (for example `ts_ms`, MySQL `file`, `pos` and `row`, or PostgreSQL `lsn` and `txId`). Fields that `metadata` already carries aren't repeated. For MySQL, `file` is the actual binlog file name, which the default output exposes under `schema`. The option is off by default, so the output is unchanged unless you enable it. +With `provide.transaction.metadata=true`, each change event has a `transaction` object (`id`, `total_order`, `data_collection_order`); otherwise `transaction` is `null`. The transaction `BEGIN` and `END` events are mapped to `status`, `id`, `ts_ms`, `event_count` and `data_collections`, where the last two are only set on `END`. + ## Quick Example The following example is how an actual setup would look like using a **MySQL Debezium Connector** and **SourceRecordJsonMapper** as the Serializable Function. From 880a3c621fa92e96d86e87d86980bc9b8e2f328c Mon Sep 17 00:00:00 2001 From: Ilyas Ahsan Date: Mon, 26 Oct 2026 21:00:47 +0700 Subject: [PATCH 5/6] [DebeziumIO] Add the transaction events information in Python and Go codebase Signed-off-by: Ilyas Ahsan --- sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go | 2 ++ sdks/python/apache_beam/io/debezium.py | 3 +++ 2 files changed, 5 insertions(+) diff --git a/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go b/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go index 88101aed0bc6..885d057484a3 100644 --- a/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go +++ b/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go @@ -146,6 +146,8 @@ func MaxTimeToRun(r int64) readOption { // ConnectionProperties specifies properties of the debezium connection passed as // a string with format [propertyName=property;]* +// Set provide.transaction.metadata=true to add a transaction key to records and +// emit transaction BEGIN/END records. func ConnectionProperties(cp []string) readOption { return func(cfg *debeziumConfig) { cfg.readSchema.ConnectionProperties = cp diff --git a/sdks/python/apache_beam/io/debezium.py b/sdks/python/apache_beam/io/debezium.py index 8e7b88da687d..032a4cdde60a 100644 --- a/sdks/python/apache_beam/io/debezium.py +++ b/sdks/python/apache_beam/io/debezium.py @@ -149,6 +149,9 @@ def __init__( :param connection_properties: properties of the debezium connection passed as string with format [propertyName=property;]* + Set ``provide.transaction.metadata=true`` + to add a ``transaction`` key to records + and emit transaction BEGIN/END records. :param start_offset: starting offset to resume the connector from a previously seen position. Provided as a list of "key=value" strings, where numeric values are From 7b3100d7bfa54ae4e2ff6dee28eb388e18b1f765 Mon Sep 17 00:00:00 2001 From: Ilyas Ahsan Date: Mon, 26 Oct 2026 21:40:57 +0700 Subject: [PATCH 6/6] [DebeziumIO] Increase the number for the new changes in DebeziumIO Signed-off-by: Ilyas Ahsan --- .github/trigger_files/beam_PostCommit_Python.json | 2 +- .github/trigger_files/beam_PostCommit_XVR_Direct.json | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index 15dcafe6956f..eed1fa8593ff 100644 --- a/.github/trigger_files/beam_PostCommit_Python.json +++ b/.github/trigger_files/beam_PostCommit_Python.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", "pr": "40241", - "modification": 66 + "modification": 67 } diff --git a/.github/trigger_files/beam_PostCommit_XVR_Direct.json b/.github/trigger_files/beam_PostCommit_XVR_Direct.json index 49cff1ed291e..6a84ff06e55f 100644 --- a/.github/trigger_files/beam_PostCommit_XVR_Direct.json +++ b/.github/trigger_files/beam_PostCommit_XVR_Direct.json @@ -1,4 +1,4 @@ { - "modification": 3, + "modification": 4, "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" }