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/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/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/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..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 @@ -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,50 @@ 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 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/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) 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):