This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 9811046f33a [DebeziumIO] Add opt-in full metadata (op and Debezium
source fields) to readAsJson (#40702)
9811046f33a is described below
commit 9811046f33aca9de59ef389290f7229ca476a7e9
Author: Tobias Kaymak <[email protected]>
AuthorDate: Thu Oct 22 17:19:05 2026 +0200
[DebeziumIO] Add opt-in full metadata (op and Debezium source fields) to
readAsJson (#40702)
* [DebeziumIO] Add opt-in full metadata (op and Debezium source fields) to
readAsJson (#40701)
---
.github/trigger_files/beam_PostCommit_Python.json | 2 +-
.../trigger_files/beam_PostCommit_XVR_Direct.json | 2 +-
sdks/go/pkg/beam/io/xlang/debeziumio/debezium.go | 17 +++
.../pkg/beam/io/xlang/debeziumio/debezium_test.go | 29 +++++
sdks/java/io/debezium/src/README.md | 2 +
.../org/apache/beam/io/debezium/DebeziumIO.java | 42 ++++++++
.../io/debezium/DebeziumTransformRegistrar.java | 9 ++
.../beam/io/debezium/KafkaSourceConsumerFn.java | 2 +-
.../apache/beam/io/debezium/SourceRecordJson.java | 119 ++++++++++++++++++++-
.../io/debezium/DebeziumIOMySqlConnectorIT.java | 74 +++++++++++++
.../apache/beam/io/debezium/DebeziumIOTest.java | 46 ++++++++
.../beam/io/debezium/SourceRecordJsonTest.java | 89 +++++++++++++++
sdks/python/apache_beam/io/debezium.py | 11 +-
13 files changed, 435 insertions(+), 9 deletions(-)
diff --git a/.github/trigger_files/beam_PostCommit_Python.json
b/.github/trigger_files/beam_PostCommit_Python.json
index b6d790320d0..15dcafe6956 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 c04f8e34f87..49cff1ed291 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 f31a71fbc9d..88101aed0bc 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 00000000000..165df74f77d
--- /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 677f42b59b0..cd5e4ceabac 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 [...]
+
## 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 f12f9a95ded..cc31c451e9c 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 class DebeziumIO {
abstract @Nullable OffsetRetainer getOffsetRetainer();
+ abstract @Nullable Boolean getFullMetadata();
+
abstract @Nullable Coder<T> getCoder();
abstract Builder<T> toBuilder();
@@ -169,6 +171,8 @@ public class DebeziumIO {
abstract Builder<T> setOffsetRetainer(OffsetRetainer retainer);
+ abstract Builder<T> setFullMetadata(Boolean fullMetadata);
+
abstract Read<T> build();
}
@@ -305,6 +309,44 @@ public class DebeziumIO {
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.
+ *
+ * <p>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<T> 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<T> resolveFormatFunction() {
+ SourceRecordMapper<T> formatFunction = getFormatFunction();
+ if (!Boolean.TRUE.equals(getFullMetadata())) {
+ return formatFunction;
+ }
+ if (formatFunction instanceof SourceRecordJson.SourceRecordJsonMapper) {
+ @SuppressWarnings("unchecked")
+ SourceRecordMapper<T> fullMetadataMapper =
+ (SourceRecordMapper<T>) new
SourceRecordJson.SourceRecordJsonMapper(true);
+ return fullMetadataMapper;
+ }
+ throw new IllegalArgumentException(
+ "withFullMetadata(true) only applies to readAsJson() /
SourceRecordJsonMapper");
+ }
+
protected Schema getRecordSchema() {
KafkaSourceConsumerFn<T> 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 4b431ac38ba..b236935814a 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 class DebeziumTransformRegistrar implements
ExternalTransformRegistrar {
private @Nullable Long maxTimeToRun;
private @Nullable List<String> startOffset;
private @Nullable String offsetStoragePath;
+ private @Nullable Boolean fullMetadata;
public void setConnectionProperties(@Nullable List<String>
connectionProperties) {
this.connectionProperties = connectionProperties;
@@ -101,6 +102,10 @@ public class DebeziumTransformRegistrar implements
ExternalTransformRegistrar {
public void setOffsetStoragePath(@Nullable String offsetStoragePath) {
this.offsetStoragePath = offsetStoragePath;
}
+
+ public void setFullMetadata(@Nullable Boolean fullMetadata) {
+ this.fullMetadata = fullMetadata;
+ }
}
@Override
@@ -161,6 +166,10 @@ public class DebeziumTransformRegistrar implements
ExternalTransformRegistrar {
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 89fc2a5c085..c9daf052de5 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<T> extends
DoFn<Map<String, String>, T> {
// this(connectorClass, fn, maxRecords, null);
this.connectorClass = (Class<? extends SourceConnector>) 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 43b03605b55..694b232d464 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 @@ package org.apache.beam.io.debezium;
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 @@ import org.checkerframework.checker.nullness.qual.Nullable;
* <li>after - {@link #loadAfter}
* <li>metadata - {@link #loadMetadata}
* <ul>
- * <li>schema - Database Schema
* <li>connector - Connector used
* <li>version - Connector version
+ * <li>name - Connector name
+ * <li>database - Database name
+ * <li>schema - Database Schema (for MySQL, the binlog file name)
+ * <li>table - Table name
* </ul>
* </ul>
*
+ * <h3>Full metadata</h3>
+ *
+ * <p>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}:
+ *
+ * <ul>
+ * <li>op - The Debezium operation ({@code c}, {@code u}, {@code d}, {@code
r}, {@code t}), if the
+ * record envelope has an {@code op} field
+ * <li>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})
+ * </ul>
+ *
+ * <p>By default ({@code fullMetadata = false}) the output only contains the
six keys above.
+ *
* <h3>Usage Example</h3>
*
* <p>Map each SourceRecord to JSON
@@ -68,8 +93,16 @@ import org.checkerframework.checker.nullness.qual.Nullable;
+ " 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<String> 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<String, Object> 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 class SourceRecordJson {
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 @@ public class SourceRecordJson {
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<String, Object> loadMetadataExtras() {
+ Struct source = (Struct) this.value.get("source");
+ Map<String, Object> 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 @@ public class SourceRecordJson {
* @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<String> {
+ 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 class SourceRecordJson {
* @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<String, Object> 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<String, Object> 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 32edc4eb56b..29c8e33ac06 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 @@ package org.apache.beam.io.debezium;
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.ResultSet;
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.PCollection;
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 class DebeziumIOMySqlConnectorIT {
p.run().waitUntilFinish();
}
+
+ /**
+ * Debezium - MySQL connector Test with {@link
DebeziumIO.Read#withFullMetadata(boolean)}.
+ *
+ * <p>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<String> 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<String> res) -> {
+ List<String> 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 074f0ba5b52..294b709dced 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 @@ package org.apache.beam.io.debezium;
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 java.util.Map;
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 class DebeziumIOTest implements Serializable {
assertEquals(explicitOffset, restriction.offset);
}
+ @Test
+ public void testReadAsJsonWithFullMetadata() throws Exception {
+ SourceRecord record =
+
SourceRecordJsonTest.buildEnvelopeRecord(SourceRecordJsonTest.postgresSource(),
"u");
+ SourceRecordMapper<String> fn =
+ DebeziumIO.readAsJson().withFullMetadata(true).resolveFormatFunction();
+ assertEquals(new SourceRecordJson(record, true).toJson(),
fn.mapSourceRecord(record));
+ }
+
+ @Test
+ public void testReadAsJsonWithoutFullMetadata() {
+ DebeziumIO.Read<String> read = DebeziumIO.readAsJson();
+ assertSame(read.getFormatFunction(), read.resolveFormatFunction());
+ }
+
+ @Test
+ public void testFullMetadataWithCustomFormatFunctionThrows() {
+ DebeziumIO.Read<String> read =
+ DebeziumIO.<String>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<String> read =
+ (DebeziumIO.Read<String>)
+ 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 add8844f48f..bcb33e97e4d 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 class SourceRecordJsonTest implements Serializable {
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 5b5efe86fba..8e7b88da687 100644
--- a/sdks/python/apache_beam/io/debezium.py
+++ b/sdks/python/apache_beam/io/debezium.py
@@ -110,7 +110,7 @@ ReadFromDebeziumSchema = NamedTuple(
('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 @@ class ReadFromDebezium(PTransform):
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 @@ class ReadFromDebezium(PTransform):
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 @@ class ReadFromDebezium(PTransform):
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):
