From 08fc3dd2ec18ba48cee4b34b20549ab239449639 Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Thu, 22 Oct 2026 07:25:30 +0000 Subject: [PATCH 1/2] [DebeziumIO] Add opt-in full metadata (op and Debezium source fields) to readAsJson (#40701) SourceRecordJson only copies six fields of the Debezium `source` struct into `metadata` (connector, version, name, database, schema, table), so values such as MySQL `file`, `pos`, `row` and `ts_ms`, PostgreSQL `lsn` and `txId`, and the change event's `op` are dropped. Add `DebeziumIO.Read#withFullMetadata(boolean)`, off by default. With the option off, the JSON output is byte for byte identical to before. With it on, `metadata` additionally gets, after `table`: - `op`, if the record envelope has an `op` field - every other field of the `source` struct in schema order, skipping those `metadata` already carries (connector, version, name, db, schema, table) so nothing is duplicated and no key is overwritten. The fields are flattened into `metadata`, there is no nested object. For MySQL this includes `file`, the real binlog file name that the default output exposes under `schema`. The option only applies to readAsJson() / SourceRecordJsonMapper; with a custom format function it fails with an IllegalArgumentException when the transform is expanded. SourceRecordJson and SourceRecordJsonMapper get boolean constructor overloads, and the existing constructors keep the default behaviour. This follows the review on #40702: Abacn suggested making the extra metadata a transform option that is off by default and tested in one case, and Ilyas Ahsan asked for `op` to count, route and soft delete events per operation. --- sdks/java/io/debezium/src/README.md | 2 + .../apache/beam/io/debezium/DebeziumIO.java | 42 +++++++ .../io/debezium/KafkaSourceConsumerFn.java | 2 +- .../beam/io/debezium/SourceRecordJson.java | 119 +++++++++++++++++- .../debezium/DebeziumIOMySqlConnectorIT.java | 74 +++++++++++ .../beam/io/debezium/DebeziumIOTest.java | 26 ++++ .../io/debezium/SourceRecordJsonTest.java | 89 +++++++++++++ 7 files changed, 349 insertions(+), 5 deletions(-) diff --git a/sdks/java/io/debezium/src/README.md b/sdks/java/io/debezium/src/README.md index 677f42b59b02..cd5e4ceabacb 100644 --- a/sdks/java/io/debezium/src/README.md +++ b/sdks/java/io/debezium/src/README.md @@ -93,6 +93,8 @@ If you want to use this built-in JSON Mapper, you can do it by setting an instan ``` > **Note:** `SourceRecordJsonMapper`comes out of the box, but you may use any Format Function you prefer. +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. + ## Quick Example The following example is how an actual setup would look like using a **MySQL Debezium Connector** and **SourceRecordJsonMapper** as the Serializable Function. 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 f12f9a95dedb..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 @@ -147,6 +147,8 @@ public abstract static class Read extends PTransform> abstract @Nullable OffsetRetainer getOffsetRetainer(); + abstract @Nullable Boolean getFullMetadata(); + abstract @Nullable Coder getCoder(); abstract Builder toBuilder(); @@ -169,6 +171,8 @@ abstract static class Builder { abstract Builder setOffsetRetainer(OffsetRetainer retainer); + abstract Builder setFullMetadata(Boolean fullMetadata); + abstract Read build(); } @@ -305,6 +309,44 @@ public Read withOffsetRetainer(OffsetRetainer retainer) { return toBuilder().setOffsetRetainer(retainer).build(); } + /** + * 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. + * + *

Only applies to {@link DebeziumIO#readAsJson()} / {@link + * SourceRecordJson.SourceRecordJsonMapper}; combining it with a custom {@link + * #withFormatFunction(SourceRecordMapper)} fails with an {@link IllegalArgumentException} when + * the transform is expanded. Default false. + * + * @param fullMetadata Whether to emit the full Debezium metadata. + * @return PTransform {@link #read} + */ + public Read withFullMetadata(boolean fullMetadata) { + return toBuilder().setFullMetadata(fullMetadata).build(); + } + + /** + * Returns the format function to use, taking {@link #withFullMetadata(boolean)} into account. + * + * @throws IllegalArgumentException if full metadata is requested but the format function is not + * a {@link SourceRecordJson.SourceRecordJsonMapper}. + */ + SourceRecordMapper resolveFormatFunction() { + SourceRecordMapper formatFunction = getFormatFunction(); + if (!Boolean.TRUE.equals(getFullMetadata())) { + return formatFunction; + } + if (formatFunction instanceof SourceRecordJson.SourceRecordJsonMapper) { + @SuppressWarnings("unchecked") + SourceRecordMapper fullMetadataMapper = + (SourceRecordMapper) new SourceRecordJson.SourceRecordJsonMapper(true); + return fullMetadataMapper; + } + throw new IllegalArgumentException( + "withFullMetadata(true) only applies to readAsJson() / SourceRecordJsonMapper"); + } + protected Schema getRecordSchema() { KafkaSourceConsumerFn fn = new KafkaSourceConsumerFn<>(getConnectorConfiguration().getConnectorClass().get(), this); diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java index 89fc2a5c085d..c9daf052de58 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/KafkaSourceConsumerFn.java @@ -113,7 +113,7 @@ public class KafkaSourceConsumerFn extends DoFn, T> { // this(connectorClass, fn, maxRecords, null); this.connectorClass = (Class) connectorClass; this.spec = spec; - this.fn = spec.getFormatFunction(); + this.fn = spec.resolveFormatFunction(); this.pollingTimeOut = MoreObjects.firstNonNull(spec.getPollingTimeout(), DEFAULT_POLLING_TIMEOUT); } 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 43b03605b559..694b232d4647 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,10 +19,15 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.Serializable; +import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; +import java.util.Set; 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.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet; import org.apache.kafka.connect.data.Field; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.errors.DataException; @@ -45,12 +50,32 @@ *

  • after - {@link #loadAfter} *
  • metadata - {@link #loadMetadata} *
      - *
    • schema - Database Schema *
    • connector - Connector used *
    • version - Connector version + *
    • name - Connector name + *
    • database - Database name + *
    • schema - Database Schema (for MySQL, the binlog file name) + *
    • table - Table name *
    * * + *

    Full metadata

    + * + *

    When constructed with {@code fullMetadata = true} (see {@link + * SourceRecordJsonMapper#SourceRecordJsonMapper(boolean)} and {@link + * DebeziumIO.Read#withFullMetadata(boolean)}), extra keys are appended to {@code metadata} after + * {@code table}: + * + *

      + *
    • op - The Debezium operation ({@code c}, {@code u}, {@code d}, {@code r}, {@code t}), if the + * record envelope has an {@code op} field + *
    • 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}) + *
    + * + *

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

    Usage Example

    * *

    Map each SourceRecord to JSON @@ -68,8 +93,16 @@ + " Making the class final or moving the throwing code" + " into a static factory method may fix it.") public class SourceRecordJson { + /** + * Source fields that are already represented in {@link Event.Metadata} (or would collide with an + * 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"); + private final @Nullable Struct value; private final @Nullable Event event; + private final Map metadataExtras; /** * Initializer. @@ -77,6 +110,17 @@ public class SourceRecordJson { * @param sourceRecord retrieved SourceRecord using a supported SourceConnector */ public SourceRecordJson(@Nullable SourceRecord sourceRecord) { + this(sourceRecord, false); + } + + /** + * Initializer. + * + * @param sourceRecord retrieved SourceRecord using a supported SourceConnector + * @param fullMetadata whether to add {@code op} and the remaining Debezium {@code source} fields + * to {@code metadata} + */ + public SourceRecordJson(@Nullable SourceRecord sourceRecord, boolean fullMetadata) { if (sourceRecord == null) { throw new IllegalArgumentException(); } @@ -85,12 +129,15 @@ public SourceRecordJson(@Nullable SourceRecord sourceRecord) { if (this.value == null) { this.event = new Event(null, null, null); + this.metadataExtras = Collections.emptyMap(); } else { Event.Metadata metadata = this.loadMetadata(); Event.Before before = this.loadBefore(); Event.After after = this.loadAfter(); this.event = new Event(metadata, before, after); + this.metadataExtras = + fullMetadata && metadata != null ? this.loadMetadataExtras() : Collections.emptyMap(); } } @@ -129,6 +176,33 @@ private Event.Metadata loadMetadata() { source.getString("table")); } + /** + * 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}. + * + * @return extra metadata keys, in insertion order + */ + private Map loadMetadataExtras() { + Struct source = (Struct) this.value.get("source"); + Map extras = new LinkedHashMap<>(); + if (this.value.schema().field("op") != null) { + extras.put("op", toJsonFriendly(this.value.get("op"))); + } + for (Field field : source.schema().fields()) { + String fieldName = field.name(); + if (!RESERVED_METADATA_KEYS.contains(fieldName)) { + extras.put(fieldName, toJsonFriendly(source.get(field))); + } + } + return extras; + } + + private static @Nullable Object toJsonFriendly(@Nullable Object value) { + // Source fields are flat primitives today; stringify a nested Struct to keep metadata flat. + return value instanceof Struct ? String.valueOf(value) : value; + } + /** * Extracts the before field within SourceRecord. * @@ -183,14 +257,31 @@ private Event.After loadAfter() { * @return JSON String */ public String toJson() { - return this.event.toJson(); + return this.event.toJson(this.metadataExtras); } /** {@link SourceRecordJson} implementation. */ public static class SourceRecordJsonMapper implements SourceRecordMapper { + private final boolean fullMetadata; + + /** Creates a mapper that emits the default metadata. */ + public SourceRecordJsonMapper() { + this(false); + } + + /** + * Creates a mapper. + * + * @param fullMetadata whether to add {@code op} and the remaining Debezium {@code source} + * fields to {@code metadata} + */ + public SourceRecordJsonMapper(boolean fullMetadata) { + this.fullMetadata = fullMetadata; + } + @Override public String mapSourceRecord(SourceRecord sourceRecord) throws Exception { - return new SourceRecordJson(sourceRecord).toJson(); + return new SourceRecordJson(sourceRecord, fullMetadata).toJson(); } } @@ -223,8 +314,28 @@ public Event( * @return JSON String */ public String toJson() { + return toJson(Collections.emptyMap()); + } + + /** + * Transforms the Event to a JSON string, appending the given extra keys to {@code metadata} + * after {@code table}. + * + * @param extras Additional metadata keys, in iteration order + * @return JSON String + */ + String toJson(Map extras) { Gson gson = new GsonBuilder().serializeNulls().create(); - return gson.toJson(this); + if (extras.isEmpty() || metadata == null) { + 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 metadataJson = tree.getAsJsonObject("metadata"); + for (Map.Entry extra : extras.entrySet()) { + metadataJson.add(extra.getKey(), gson.toJsonTree(extra.getValue())); + } + return gson.toJson(tree); } /** Depicts the metadata within a SourceRecord. It has valuable fields. */ 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 32edc4eb56b4..29c8e33ac06d 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 @@ -20,7 +20,11 @@ import static org.apache.beam.io.debezium.DebeziumIOPostgresSqlConnectorIT.TABLE_SCHEMA; import static org.apache.beam.sdk.testing.SerializableMatchers.hasItem; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.empty; import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.not; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import com.zaxxer.hikari.HikariConfig; @@ -31,6 +35,7 @@ import java.sql.SQLException; import java.sql.Statement; import java.time.Duration; +import java.util.List; import javax.sql.DataSource; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.coders.StringUtf8Coder; @@ -44,6 +49,8 @@ import org.apache.beam.sdk.values.PCollectionRowTuple; import org.apache.beam.sdk.values.Row; import org.apache.beam.sdk.values.TypeDescriptors; +import org.apache.beam.vendor.grpc.v1p69p0.com.google.gson.JsonObject; +import org.apache.beam.vendor.grpc.v1p69p0.com.google.gson.JsonParser; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.junit.ClassRule; import org.junit.Test; @@ -293,4 +300,71 @@ public void testDebeziumIOMySql() { p.run().waitUntilFinish(); } + + /** + * Debezium - MySQL connector Test with {@link DebeziumIO.Read#withFullMetadata(boolean)}. + * + *

    Tests that the JSON metadata carries {@code op} and the MySQL source fields, flattened. + */ + @Test + public void testDebeziumIOMySqlFullMetadata() { + + String kafkaBootstrapServers = KAFKA_CONTAINER.getBootstrapServers(); + String schemaHistoryTopic = "mysql-schema-history-full-metadata-" + System.nanoTime(); + + String host = MY_SQL_CONTAINER.getHost(); + String port = MY_SQL_CONTAINER.getMappedPort(3306).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(MySqlConnector.class) + .withHostName(host) + .withPort(port) + .withConnectionProperty("database.server.id", "184055") + .withConnectionProperty("database.include.list", "inventory") + .withConnectionProperty("include.schema.changes", "false") + .withConnectionProperty( + "schema.history.internal.kafka.bootstrap.servers", + kafkaBootstrapServers) + .withConnectionProperty( + "schema.history.internal.kafka.topic", schemaHistoryTopic) + .withConnectionProperty( + "schema.history.internal", + "io.debezium.storage.kafka.history.KafkaSchemaHistory") + .withConnectionProperty( + "schema.history.internal.store.only.captured.tables.ddl", "true")) + .withFullMetadata(true) + .withMaxNumberOfRecords(10)); + + PAssert.that(results) + .satisfies( + (Iterable res) -> { + List records = Lists.newArrayList(res); + assertThat(records, not(empty())); + for (String record : records) { + JsonObject metadata = + JsonParser.parseString(record).getAsJsonObject().getAsJsonObject("metadata"); + // Snapshot rows are reads. + assertThat(metadata.get("op").getAsString(), equalTo("r")); + assertTrue(metadata.get("pos").getAsJsonPrimitive().isNumber()); + assertTrue(metadata.get("row").getAsJsonPrimitive().isNumber()); + assertTrue(metadata.get("ts_ms").getAsJsonPrimitive().isNumber()); + // MySQL keeps exposing the binlog file under "schema" for compatibility. + assertThat( + metadata.get("file").getAsString(), + equalTo(metadata.get("schema").getAsString())); + assertFalse(metadata.has("source")); + } + return null; + }); + + p.run().waitUntilFinish(); + } } diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java index 074f0ba5b526..48d4cc5e4a87 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java +++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; @@ -33,6 +34,7 @@ import org.apache.beam.io.debezium.DebeziumIO.ConnectorConfiguration; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.kafka.common.config.ConfigValue; +import org.apache.kafka.connect.source.SourceRecord; import org.checkerframework.checker.nullness.qual.Nullable; import org.junit.Test; import org.junit.runner.RunWith; @@ -235,6 +237,30 @@ public void testRetainerFallsBackToWithStartOffsetWhenLoadReturnsNull() throws E assertEquals(explicitOffset, restriction.offset); } + @Test + public void testReadAsJsonWithFullMetadata() throws Exception { + SourceRecord record = + SourceRecordJsonTest.buildEnvelopeRecord(SourceRecordJsonTest.postgresSource(), "u"); + SourceRecordMapper fn = + DebeziumIO.readAsJson().withFullMetadata(true).resolveFormatFunction(); + assertEquals(new SourceRecordJson(record, true).toJson(), fn.mapSourceRecord(record)); + } + + @Test + public void testReadAsJsonWithoutFullMetadata() { + DebeziumIO.Read read = DebeziumIO.readAsJson(); + assertSame(read.getFormatFunction(), read.resolveFormatFunction()); + } + + @Test + public void testFullMetadataWithCustomFormatFunctionThrows() { + DebeziumIO.Read read = + DebeziumIO.read() + .withFormatFunction(sourceRecord -> sourceRecord.toString()) + .withFullMetadata(true); + assertThrows(IllegalArgumentException.class, read::resolveFormatFunction); + } + @Test public void testBuildExternalThrowsOnMalformedStartOffsetEntry() { DebeziumTransformRegistrar.ReadBuilder.Configuration config = 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 add8844f48f8..bcb33e97e4df 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 @@ -18,13 +18,16 @@ package org.apache.beam.io.debezium; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; import java.io.Serializable; import org.apache.kafka.connect.data.Schema; import org.apache.kafka.connect.data.SchemaBuilder; import org.apache.kafka.connect.data.Struct; import org.apache.kafka.connect.source.SourceRecord; +import org.checkerframework.checker.nullness.qual.Nullable; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -58,6 +61,92 @@ public void testSourceRecordJsonWhenSourceRecordIsNull() { assertThrows(IllegalArgumentException.class, () -> new SourceRecordJson(null)); } + @Test + public void testFullMetadata() { + // MySQL has no "schema" source field, so the binlog "file" is also exposed as "schema". + Schema mySqlSchema = + sourceSchemaBuilder() + .field("gtid", Schema.OPTIONAL_STRING_SCHEMA) + .field("file", Schema.STRING_SCHEMA) + .field("pos", Schema.INT64_SCHEMA) + .field("row", Schema.INT32_SCHEMA) + .build(); + Struct mySql = + sourceStruct(mySqlSchema, "mysql", "inventory") + .put("file", "mysql-bin.000003") + .put("pos", 796L) + .put("row", 0); + assertEquals( + "{\"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}}}", + new SourceRecordJson(buildEnvelopeRecord(mySql, "c"), true).toJson()); + + // PostgreSQL "schema" is a source field: it stays the existing key and isn't repeated. + assertEquals( + "{\"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}}}", + new SourceRecordJson(buildEnvelopeRecord(postgresSource(), "u"), true).toJson()); + } + + @Test + public void testFullMetadataWithoutOpField() { + String json = new SourceRecordJson(buildEnvelopeRecord(postgresSource(), null), true).toJson(); + assertFalse(json.contains("\"op\"")); + assertTrue(json.contains("\"lsn\":37094192")); + } + + private static SchemaBuilder sourceSchemaBuilder() { + return SchemaBuilder.struct() + .field("version", Schema.STRING_SCHEMA) + .field("connector", Schema.STRING_SCHEMA) + .field("name", Schema.STRING_SCHEMA) + .field("ts_ms", Schema.INT64_SCHEMA) + .field("db", Schema.STRING_SCHEMA) + .field("table", Schema.STRING_SCHEMA); + } + + private static Struct sourceStruct(Schema schema, String connector, String db) { + return new Struct(schema) + .put("version", "3.5.2.Final") + .put("connector", connector) + .put("name", "n") + .put("ts_ms", 1791689147000L) + .put("db", db) + .put("table", "customers"); + } + + static Struct postgresSource() { + Schema schema = + sourceSchemaBuilder() + .field("schema", Schema.STRING_SCHEMA) + .field("lsn", Schema.INT64_SCHEMA) + .build(); + return sourceStruct(schema, "postgresql", "postgres") + .put("schema", "inventory") + .put("lsn", 37094192L); + } + + /** A Debezium-like envelope with {@code before} null, {@code after.id} and an optional op. */ + static SourceRecord buildEnvelopeRecord(Struct source, @Nullable String op) { + Schema row = SchemaBuilder.struct().optional().field("id", Schema.INT32_SCHEMA).build(); + SchemaBuilder envelope = + SchemaBuilder.struct() + .field("before", row) + .field("after", row) + .field("source", source.schema()); + Schema schema = (op == null ? envelope : envelope.field("op", Schema.STRING_SCHEMA)).build(); + Struct value = + new Struct(schema).put("after", new Struct(row).put("id", 1001)).put("source", source); + if (op != null) { + value.put("op", op); + } + return new SourceRecord(null, null, null, schema, value); + } + private static Schema buildSourceSchema() { return SchemaBuilder.struct() .field("connector", Schema.STRING_SCHEMA) From a28e768195b26202eb4c4cffe8b5276c93e4073b Mon Sep 17 00:00:00 2001 From: Tobias Kaymak Date: Thu, 22 Oct 2026 07:25:31 +0000 Subject: [PATCH 2/2] [DebeziumIO] Expose full_metadata option in Python and Go ReadFromDebezium (#40701) Wire the new DebeziumIO.Read#withFullMetadata option through the cross-language registrar (`setFullMetadata`, boxed and nullable like the other optional settings; only an explicit true enables it), the Python `ReadFromDebezium(full_metadata=...)` keyword argument (appended at the end of ReadFromDebeziumSchema) and the Go `debeziumio.FullMetadata()` read option. The default output is unchanged, so the Python and Go xlang ITs stay as they are. Touch the PostCommit Python and PostCommit XVR Direct trigger files because those suites run the Python and Go Debezium xlang tests. --- .../trigger_files/beam_PostCommit_Python.json | 2 +- .../beam_PostCommit_XVR_Direct.json | 2 +- .../pkg/beam/io/xlang/debeziumio/debezium.go | 17 +++++++++++ .../beam/io/xlang/debeziumio/debezium_test.go | 29 +++++++++++++++++++ .../debezium/DebeziumTransformRegistrar.java | 9 ++++++ .../beam/io/debezium/DebeziumIOTest.java | 20 +++++++++++++ sdks/python/apache_beam/io/debezium.py | 11 +++++-- 7 files changed, 86 insertions(+), 4 deletions(-) create mode 100644 sdks/go/pkg/beam/io/xlang/debeziumio/debezium_test.go diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index b6d790320d0b..15dcafe6956f 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": 65 + "modification": 66 } diff --git a/.github/trigger_files/beam_PostCommit_XVR_Direct.json b/.github/trigger_files/beam_PostCommit_XVR_Direct.json index c04f8e34f87f..49cff1ed291e 100644 --- a/.github/trigger_files/beam_PostCommit_XVR_Direct.json +++ b/.github/trigger_files/beam_PostCommit_XVR_Direct.json @@ -1,4 +1,4 @@ { - "modification": 2, + "modification": 3, "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" } diff --git a/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go b/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go index f31a71fbc9d9..88101aed0bc6 100644 --- a/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go +++ b/sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go @@ -79,6 +79,7 @@ type readFromDebeziumSchema struct { ConnectionProperties []string StartOffset []string OffsetStoragePath *string + FullMetadata *bool } type debeziumConfig struct { @@ -189,6 +190,22 @@ func OffsetStoragePath(path string) readOption { } } +// FullMetadata adds op and all Debezium source fields not already present +// (e.g. MySQL file, pos, row, ts_ms; PostgreSQL lsn, txId, ts_ms) to the +// metadata of each JSON record. By default, only connector, version, name, +// database, schema and table are included. +// +// Example: +// +// debeziumio.Read(s, username, password, host, port, debeziumio.MySQL, +// reflectx.String, debeziumio.FullMetadata()) +func FullMetadata() readOption { + return func(cfg *debeziumConfig) { + b := true + cfg.readSchema.FullMetadata = &b + } +} + // ExpansionAddr sets the expansion service address to use for DebeziumIO cross-langauage transform. func ExpansionAddr(expansionAddr string) readOption { return func(cfg *debeziumConfig) { diff --git a/sdks/go/pkg/beam/io/xlang/debeziumio/debezium_test.go b/sdks/go/pkg/beam/io/xlang/debeziumio/debezium_test.go new file mode 100644 index 000000000000..165df74f77db --- /dev/null +++ b/sdks/go/pkg/beam/io/xlang/debeziumio/debezium_test.go @@ -0,0 +1,29 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package debeziumio + +import "testing" + +func TestFullMetadata(t *testing.T) { + dc := debeziumConfig{readSchema: &readFromDebeziumSchema{}} + if dc.readSchema.FullMetadata != nil { + t.Fatalf("FullMetadata = %v by default, want nil", *dc.readSchema.FullMetadata) + } + FullMetadata()(&dc) + if got := dc.readSchema.FullMetadata; got == nil || !*got { + t.Errorf("FullMetadata() set FullMetadata = %v, want true", got) + } +} diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java index 4b431ac38ba2..b236935814ab 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java @@ -81,6 +81,7 @@ public static class Configuration extends CrossLanguageConfiguration { private @Nullable Long maxTimeToRun; private @Nullable List startOffset; private @Nullable String offsetStoragePath; + private @Nullable Boolean fullMetadata; public void setConnectionProperties(@Nullable List connectionProperties) { this.connectionProperties = connectionProperties; @@ -101,6 +102,10 @@ public void setStartOffset(@Nullable List startOffset) { public void setOffsetStoragePath(@Nullable String offsetStoragePath) { this.offsetStoragePath = offsetStoragePath; } + + public void setFullMetadata(@Nullable Boolean fullMetadata) { + this.fullMetadata = fullMetadata; + } } @Override @@ -161,6 +166,10 @@ public PTransform> buildExternal(Configuration confi FileSystemOffsetRetainer.of(configuration.offsetStoragePath)); } + if (Boolean.TRUE.equals(configuration.fullMetadata)) { + readTransform = readTransform.withFullMetadata(true); + } + return readTransform; } } diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java index 48d4cc5e4a87..294b709dcedc 100644 --- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java +++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java @@ -261,6 +261,26 @@ public void testFullMetadataWithCustomFormatFunctionThrows() { assertThrows(IllegalArgumentException.class, read::resolveFormatFunction); } + @Test + public void testBuildExternalWithFullMetadata() throws Exception { + DebeziumTransformRegistrar.ReadBuilder.Configuration config = + new DebeziumTransformRegistrar.ReadBuilder.Configuration(); + config.setUsername("user"); + config.setPassword("pass"); + config.setHost("localhost"); + config.setPort("3306"); + config.setConnectorClass("MySQL"); + config.setFullMetadata(true); + DebeziumIO.Read read = + (DebeziumIO.Read) + new DebeziumTransformRegistrar.ReadBuilder().buildExternal(config); + SourceRecord record = + SourceRecordJsonTest.buildEnvelopeRecord(SourceRecordJsonTest.postgresSource(), "u"); + assertEquals( + new SourceRecordJson(record, true).toJson(), + read.resolveFormatFunction().mapSourceRecord(record)); + } + @Test public void testBuildExternalThrowsOnMalformedStartOffsetEntry() { DebeziumTransformRegistrar.ReadBuilder.Configuration config = diff --git a/sdks/python/apache_beam/io/debezium.py b/sdks/python/apache_beam/io/debezium.py index 5b5efe86fba9..8e7b88da687d 100644 --- a/sdks/python/apache_beam/io/debezium.py +++ b/sdks/python/apache_beam/io/debezium.py @@ -110,7 +110,7 @@ class DriverClassName(Enum): ('host', str), ('port', str), ('max_number_of_records', Optional[int]), ('connection_properties', list[str]), ('start_offset', Optional[list[str]]), - ('offset_storage_path', Optional[str])]) + ('offset_storage_path', Optional[str]), ('full_metadata', Optional[bool])]) class ReadFromDebezium(PTransform): @@ -134,6 +134,7 @@ def __init__( connection_properties=None, start_offset=None, offset_storage_path=None, + full_metadata=None, expansion_service=None): """ Initializes a read operation from Debezium. @@ -179,6 +180,11 @@ def __init__( When set, takes precedence over ``start_offset``. + :param full_metadata: when True, adds ``op`` and all Debezium source + fields not already present (e.g. MySQL + ``file``, ``pos``, ``row``, ``ts_ms``; + PostgreSQL ``lsn``, ``txId``, ``ts_ms``) to + ``metadata``. Defaults to False. :param expansion_service: The address (host:port) of the ExpansionService. """ @@ -192,7 +198,8 @@ def __init__( max_number_of_records=max_number_of_records, connection_properties=connection_properties, start_offset=start_offset, - offset_storage_path=offset_storage_path) + offset_storage_path=offset_storage_path, + full_metadata=full_metadata) self.expansion_service = expansion_service or default_io_expansion_service() def expand(self, pbegin):