From 621a78cd4123bf05a525027b531a633347473a6d Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Tue, 21 Jul 2026 15:47:32 +0400 Subject: [PATCH 1/7] Add Beam YAML support for DebeziumIO --- .../DebeziumReadSchemaTransformProvider.java | 76 ++++++++-- ...beziumReadSchemaTransformProviderTest.java | 139 ++++++++++++++++++ sdks/python/apache_beam/yaml/standard_io.yaml | 21 +++ 3 files changed, 220 insertions(+), 16 deletions(-) create mode 100644 sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProviderTest.java diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java index d1d5103cc496..1a3055e4bf4e 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java @@ -20,12 +20,13 @@ import com.google.auto.service.AutoService; import com.google.auto.value.AutoValue; import java.util.Arrays; -import java.util.Collection; import java.util.Collections; import java.util.List; -import java.util.stream.Collectors; import org.apache.beam.sdk.coders.RowCoder; +import org.apache.beam.sdk.schemas.AutoValueSchema; import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.schemas.annotations.DefaultSchema; +import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription; import org.apache.beam.sdk.schemas.transforms.SchemaTransform; import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider; import org.apache.beam.sdk.schemas.transforms.TypedSchemaTransformProvider; @@ -73,6 +74,19 @@ protected DebeziumReadSchemaTransformProvider( this.testLimitMilliseconds = timeLimitMs; } + private static Connectors parseConnector(String value) { + try { + return Connectors.valueOf(value); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException( + "Unsupported connector '" + + value + + "'. Supported connectors are: " + + Arrays.toString(Connectors.values()), + e); + } + } + @Override protected @NonNull @Initialized Class configurationClass() { @@ -82,21 +96,15 @@ protected DebeziumReadSchemaTransformProvider( @Override protected @NonNull @Initialized SchemaTransform from( DebeziumReadSchemaTransformConfiguration configuration) { - // TODO(pabloem): Validate configuration parameters to ensure formatting is correct. + + configuration.validate(); + Connectors connector = parseConnector(configuration.getDatabase()); + return new SchemaTransform() { @Override public PCollectionRowTuple expand(PCollectionRowTuple input) { // TODO(pabloem): Test this behavior - Collection connectors = - Arrays.stream(Connectors.values()).map(Object::toString).collect(Collectors.toSet()); - if (!connectors.contains(configuration.getDatabase())) { - throw new IllegalArgumentException( - "Unsupported database " - + configuration.getDatabase() - + ". Unable to select a JDBC driver for it. Supported Databases are: " - + String.join(", ", connectors)); - } - Class connectorClass = Connectors.valueOf(configuration.getDatabase()).getConnector(); + Class connectorClass = connector.getConnector(); DebeziumIO.ConnectorConfiguration connectorConfiguration = DebeziumIO.ConnectorConfiguration.create() .withUsername(configuration.getUsername()) @@ -123,9 +131,9 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { configuration.getDebeziumConnectionProperties(); if (debeziumConnectionProperties != null) { for (String connectionProperty : debeziumConnectionProperties) { - String[] parts = connectionProperty.split("=", -1); - String key = parts[0]; - String value = parts[1]; + int separator = connectionProperty.indexOf('='); + String key = connectionProperty.substring(0, separator); + String value = connectionProperty.substring(separator + 1); connectorConfiguration = connectorConfiguration.withConnectionProperty(key, value); } } @@ -172,22 +180,58 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { return Collections.singletonList("output"); } + @DefaultSchema(AutoValueSchema.class) @AutoValue public abstract static class DebeziumReadSchemaTransformConfiguration { + @SchemaFieldDescription("Username used to connect to the source database.") public abstract String getUsername(); + @SchemaFieldDescription("Password used to connect to the source database.") public abstract String getPassword(); + @SchemaFieldDescription("Hostname of the source database.") public abstract String getHost(); + @SchemaFieldDescription("Port of the source database.") public abstract int getPort(); + @SchemaFieldDescription("Fully qualified table name included in the Debezium change stream.") public abstract String getTable(); + @SchemaFieldDescription( + "Debezium connector type. Supported values: MYSQL, POSTGRES, SQLSERVER, ORACLE, and DB2.") public abstract @NonNull String getDatabase(); + @SchemaFieldDescription("Additional Debezium connection properties in key=value format.") public abstract @Nullable List getDebeziumConnectionProperties(); + public void validate() { + if (getHost().isEmpty()) { + throw new IllegalArgumentException("host must not be empty."); + } + + if (getPort() <= 0 || getPort() > 65535) { + throw new IllegalArgumentException( + "port must be between 1 and 65535, but was " + getPort() + "."); + } + + if (getTable().isEmpty()) { + throw new IllegalArgumentException("table must not be empty."); + } + + List connectionProperties = getDebeziumConnectionProperties(); + if (connectionProperties != null) { + for (String property : connectionProperties) { + if (property == null || property.indexOf('=') <= 0) { + throw new IllegalArgumentException( + "Invalid Debezium connection property '" + + property + + "'. Expected key=value format."); + } + } + } + } + public static Builder builder() { return new AutoValue_DebeziumReadSchemaTransformProvider_DebeziumReadSchemaTransformConfiguration .Builder(); diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProviderTest.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProviderTest.java new file mode 100644 index 000000000000..07bd91a5f522 --- /dev/null +++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProviderTest.java @@ -0,0 +1,139 @@ +/* + * 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 org.apache.beam.io.debezium; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.Assert.assertThrows; + +import java.util.Arrays; +import java.util.Collections; +import org.hamcrest.Matchers; +import org.junit.Test; + +public class DebeziumReadSchemaTransformProviderTest { + + private static DebeziumReadSchemaTransformProvider.DebeziumReadSchemaTransformConfiguration + .Builder + validConfiguration() { + return DebeziumReadSchemaTransformProvider.DebeziumReadSchemaTransformConfiguration.builder() + .setUsername("user") + .setPassword("password") + .setHost("localhost") + .setPort(5432) + .setDatabase("POSTGRES") + .setTable("inventory.customers") + .setDebeziumConnectionProperties(Collections.emptyList()); + } + + @Test + public void testValidConfiguration() { + validConfiguration().build().validate(); + } + + @Test + public void testEmptyHost() { + IllegalArgumentException exception = + assertThrows( + IllegalArgumentException.class, + () -> validConfiguration().setHost("").build().validate()); + + assertThat(exception.getMessage(), Matchers.containsString("host must not be empty")); + } + + @Test + public void testInvalidPort() { + IllegalArgumentException exception = + assertThrows( + IllegalArgumentException.class, + () -> validConfiguration().setPort(0).build().validate()); + + assertThat(exception.getMessage(), Matchers.containsString("port must be between")); + } + + @Test + public void testEmptyTable() { + IllegalArgumentException exception = + assertThrows( + IllegalArgumentException.class, + () -> validConfiguration().setTable("").build().validate()); + + assertThat(exception.getMessage(), Matchers.containsString("table must not be empty")); + } + + @Test + public void testInvalidConnectionProperty() { + IllegalArgumentException exception = + assertThrows( + IllegalArgumentException.class, + () -> + validConfiguration() + .setDebeziumConnectionProperties(Collections.singletonList("invalid-property")) + .build() + .validate()); + + assertThat(exception.getMessage(), Matchers.containsString("Expected key=value format")); + } + + @Test + public void testNullConnectionProperty() { + IllegalArgumentException exception = + assertThrows( + IllegalArgumentException.class, + () -> + validConfiguration() + .setDebeziumConnectionProperties(Arrays.asList("valid=value", null)) + .build() + .validate()); + + assertThat(exception.getMessage(), Matchers.containsString("Expected key=value format")); + } + + @Test + public void testConnectionPropertyValueContainingEquals() { + validConfiguration() + .setDebeziumConnectionProperties( + Collections.singletonList("some.property=value=containing=equals")) + .build() + .validate(); + } + + @Test + public void testUnsupportedConnector() { + IllegalArgumentException exception = + assertThrows( + IllegalArgumentException.class, + () -> + new DebeziumReadSchemaTransformProvider() + .from(validConfiguration().setDatabase("UNKNOWN").build())); + + assertThat(exception.getMessage(), Matchers.containsString("Unsupported connector 'UNKNOWN'")); + assertThat(exception.getMessage(), Matchers.containsString("MYSQL")); + assertThat(exception.getMessage(), Matchers.containsString("POSTGRES")); + } + + @Test + public void testProviderContract() { + DebeziumReadSchemaTransformProvider provider = new DebeziumReadSchemaTransformProvider(); + + assertThat(provider.inputCollectionNames(), Matchers.empty()); + assertThat(provider.outputCollectionNames(), Matchers.contains("output")); + assertThat( + provider.identifier(), + Matchers.equalTo("beam:schematransform:org.apache.beam:debezium_read:v1")); + } +} diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml index 8c074e0885dd..6ee925ba9533 100644 --- a/sdks/python/apache_beam/yaml/standard_io.yaml +++ b/sdks/python/apache_beam/yaml/standard_io.yaml @@ -182,6 +182,27 @@ config: gradle_target: 'sdks:java:extensions:schemaio-expansion-service:shadowJar' +# Debezium +- type: renaming + transforms: + 'ReadFromDebezium': 'ReadFromDebezium' + config: + mappings: + 'ReadFromDebezium': + connector: 'database' + username: 'username' + password: 'password' + host: 'host' + port: 'port' + table: 'table' + connection_properties: 'debezium_connection_properties' + underlying_provider: + type: beamJar + transforms: + 'ReadFromDebezium': 'beam:schematransform:org.apache.beam:debezium_read:v1' + config: + gradle_target: 'sdks:java:io:debezium:expansion-service:shadowJar' + # Databases - type: renaming transforms: From 01c668b1455c203688e29a4d8017a2e3181a1ef1 Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Thu, 23 Jul 2026 16:02:55 +0400 Subject: [PATCH 2/7] Add Debezium YAML integration test --- .../extended_tests/databases/debezium.yaml | 48 +++++++++++++ .../apache_beam/yaml/integration_tests.py | 70 ++++++++++++++++++- sdks/python/build.gradle | 4 +- 3 files changed, 120 insertions(+), 2 deletions(-) create mode 100644 sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml diff --git a/sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml b/sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml new file mode 100644 index 000000000000..c4f7e3dbdee7 --- /dev/null +++ b/sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml @@ -0,0 +1,48 @@ +# +# 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. +# + +fixtures: + - name: DEBEZIUM_DB + type: "apache_beam.yaml.integration_tests.temp_debezium_postgres_database" + +pipelines: + - pipeline: + type: chain + transforms: + - type: ReadFromDebezium + config: + connector: POSTGRES + username: "{DEBEZIUM_DB[USERNAME]}" + password: "{DEBEZIUM_DB[PASSWORD]}" + host: "{DEBEZIUM_DB[HOST]}" + port: "{DEBEZIUM_DB[PORT]}" + table: "{DEBEZIUM_DB[TABLE]}" + connection_properties: + - "database.dbname={DEBEZIUM_DB[DATABASE]}" + - "plugin.name=pgoutput" + - "snapshot.mode=initial_only" + - type: MapToFields + config: + language: python + fields: + id: "after.id" + name: "after.name" + - type: AssertEqual + config: + elements: + - {id: 1, name: Alice} + - {id: 2, name: Bob} diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 4fca20ade966..979bcda44b77 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -48,7 +48,7 @@ class BigEndianIntegerCoderImpl(CoderImpl): """Coder implementation for big-endian integers used in cross-language tests. - + This is needed because Java's BigEndianIntegerCoder falls back to the generic 'beam:coders:javasdk:0.1' URN when used in cross-language pipelines, and Python's FnApiRunner needs to know how to decode it. @@ -394,6 +394,74 @@ def temp_mysql_database(): yield jdbc_url +@contextlib.contextmanager +def temp_debezium_postgres_database(): + """Provides a temporary PostgreSQL database configured for Debezium CDC.""" + + container = ( + DockerContainer('quay.io/debezium/example-postgres:latest') + .with_env('POSTGRES_USER', 'debezium') + .with_env('POSTGRES_PASSWORD', 'dbz') + .with_env('POSTGRES_DB', 'inventory') + .with_exposed_ports(5432)) + + try: + container.start() + wait_for_logs(container, 'database system is ready to accept connections') + + host = container.get_container_host_ip() + port = int(container.get_exposed_port(5432)) + + connection = None + for _ in range(30): + try: + connection = psycopg2.connect( + host=host, + port=port, + user='debezium', + password='dbz', + dbname='inventory') + break + except psycopg2.OperationalError: + time.sleep(1) + + if connection is None: + raise RuntimeError('Debezium PostgreSQL container failed to become ready.') + + try: + with connection.cursor() as cursor: + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS customers ( + id INTEGER PRIMARY KEY, + name VARCHAR(255) + ) + """) + cursor.execute( + """ + INSERT INTO customers (id, name) + VALUES + (1, 'Alice'), + (2, 'Bob') + ON CONFLICT (id) DO NOTHING + """) + connection.commit() + finally: + connection.close() + + yield { + 'HOST': host, + 'PORT': port, + 'USERNAME': 'debezium', + 'PASSWORD': 'dbz', + 'DATABASE': 'inventory', + 'TABLE': 'public.customers', + } + + finally: + container.stop() + + @contextlib.contextmanager def temp_postgres_database(): """Context manager to provide a temporary PostgreSQL database for testing. diff --git a/sdks/python/build.gradle b/sdks/python/build.gradle index 9e2fe232c42c..b50660978b2a 100644 --- a/sdks/python/build.gradle +++ b/sdks/python/build.gradle @@ -151,6 +151,7 @@ tasks.register("yamlIntegrationTests") { dependsOn ":sdks:java:extensions:sql:expansion-service:shadowJar" dependsOn ":sdks:java:io:expansion-service:build" dependsOn ":sdks:java:io:google-cloud-platform:expansion-service:build" + dependsOn ":sdks:java:io:debezium:expansion-service:shadowJar" doLast { exec { @@ -170,6 +171,7 @@ tasks.register("postCommitYamlIntegrationTests") { dependsOn ":sdks:java:extensions:sql:expansion-service:shadowJar" dependsOn ":sdks:java:io:expansion-service:build" dependsOn ":sdks:java:io:google-cloud-platform:expansion-service:build" + dependsOn ":sdks:java:io:debezium:expansion-service:shadowJar" doLast { def testSetInput = project.findProperty('yamlTestSet') ?: 'data,databases,messaging' @@ -188,7 +190,7 @@ tasks.register("postCommitYamlIntegrationTests") { test_files_dir = 'extended_tests/messaging' break default: - throw StopExecutionException("Unknown yamlTestSet: ${testSet}. Must be one of 'data', 'databases', or 'messaging'.") + throw StopExecutionException("Unknown yamlTestSet: ${currentTestSet}. Must be one of 'data', 'databases', or 'messaging'.") } exec { executable 'sh' From a2267554a167381ca9369b6f79d6168264d18710 Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Mon, 27 Jul 2026 17:59:25 +0400 Subject: [PATCH 3/7] Fix record schema --- .../io/debezium/DebeziumReadSchemaTransformProvider.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java index 1a3055e4bf4e..88f1b82e9520 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java @@ -74,6 +74,10 @@ protected DebeziumReadSchemaTransformProvider( this.testLimitMilliseconds = timeLimitMs; } + private static Schema withoutOptions(Schema schema) { + return Schema.builder().addFields(schema.getFields()).build(); + } + private static Connectors parseConnector(String value) { try { return Connectors.valueOf(value); @@ -149,7 +153,7 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { } // TODO(pabloem): Database connection issues can be debugged here. - Schema recordSchema = readTransform.getRecordSchema(); + Schema recordSchema = withoutOptions(readTransform.getRecordSchema()); LOG.info( "Computed schema for table {} from {}: {}", configuration.getTable(), From 25ed2f8b5b7b8246ac0ada3ee5bda9f5fc6f9e47 Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Mon, 27 Jul 2026 22:04:50 +0400 Subject: [PATCH 4/7] With max num of records --- .../DebeziumReadSchemaTransformProvider.java | 20 +++++++++++++++++++ .../extended_tests/databases/debezium.yaml | 2 ++ sdks/python/apache_beam/yaml/standard_io.yaml | 2 ++ 3 files changed, 24 insertions(+) diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java index 88f1b82e9520..6a72b3ef4a6f 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java @@ -150,6 +150,18 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { readTransform .withMaxNumberOfRecords(testLimitRecords) .withMaxTimeToRun(testLimitMilliseconds); + } else { + if (configuration.getMaxNumberOfRecords() != null) { + readTransform = + readTransform.withMaxNumberOfRecords( + configuration.getMaxNumberOfRecords()); + } + + if (configuration.getMaxTimeToRun() != null) { + readTransform = + readTransform.withMaxTimeToRun( + configuration.getMaxTimeToRun()); + } } // TODO(pabloem): Database connection issues can be debugged here. @@ -187,6 +199,14 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { @DefaultSchema(AutoValueSchema.class) @AutoValue public abstract static class DebeziumReadSchemaTransformConfiguration { + @Nullable + @SchemaFieldDescription("Maximum number of records to read before stopping.") + public abstract Integer getMaxNumberOfRecords(); + + @Nullable + @SchemaFieldDescription("Maximum time in milliseconds to run before stopping.") + public abstract Long getMaxTimeToRun(); + @SchemaFieldDescription("Username used to connect to the source database.") public abstract String getUsername(); diff --git a/sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml b/sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml index c4f7e3dbdee7..b8ada54532f4 100644 --- a/sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml +++ b/sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml @@ -31,6 +31,8 @@ pipelines: host: "{DEBEZIUM_DB[HOST]}" port: "{DEBEZIUM_DB[PORT]}" table: "{DEBEZIUM_DB[TABLE]}" + max_number_of_records: 2 + max_time_to_run: 600000 connection_properties: - "database.dbname={DEBEZIUM_DB[DATABASE]}" - "plugin.name=pgoutput" diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml index 6ee925ba9533..796429b5bdfc 100644 --- a/sdks/python/apache_beam/yaml/standard_io.yaml +++ b/sdks/python/apache_beam/yaml/standard_io.yaml @@ -196,6 +196,8 @@ port: 'port' table: 'table' connection_properties: 'debezium_connection_properties' + max_number_of_records: 'max_number_of_records' + max_time_to_run: 'max_time_to_run' underlying_provider: type: beamJar transforms: From aae48ec3ff09ed9db29e143125a7c3645e82c292 Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Mon, 27 Jul 2026 22:37:36 +0400 Subject: [PATCH 5/7] Add setters --- .../debezium/DebeziumReadSchemaTransformProvider.java | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java index 6a72b3ef4a6f..2f54b59db362 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java @@ -153,14 +153,11 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { } else { if (configuration.getMaxNumberOfRecords() != null) { readTransform = - readTransform.withMaxNumberOfRecords( - configuration.getMaxNumberOfRecords()); + readTransform.withMaxNumberOfRecords(configuration.getMaxNumberOfRecords()); } if (configuration.getMaxTimeToRun() != null) { - readTransform = - readTransform.withMaxTimeToRun( - configuration.getMaxTimeToRun()); + readTransform = readTransform.withMaxTimeToRun(configuration.getMaxTimeToRun()); } } @@ -263,6 +260,10 @@ public static Builder builder() { @AutoValue.Builder public abstract static class Builder { + public abstract Builder setMaxNumberOfRecords(@Nullable Integer maxNumberOfRecords); + + public abstract Builder setMaxTimeToRun(@Nullable Long maxTimeToRun); + public abstract Builder setUsername(String username); public abstract Builder setPassword(String password); From 2541fc77cb33be87bd5a6238e1c4f53e25494368 Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Tue, 28 Jul 2026 15:41:27 +0400 Subject: [PATCH 6/7] Refactoring --- .../DebeziumReadSchemaTransformProvider.java | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java index 2f54b59db362..aac743a388dd 100644 --- a/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java +++ b/sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java @@ -151,13 +151,13 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { .withMaxNumberOfRecords(testLimitRecords) .withMaxTimeToRun(testLimitMilliseconds); } else { - if (configuration.getMaxNumberOfRecords() != null) { - readTransform = - readTransform.withMaxNumberOfRecords(configuration.getMaxNumberOfRecords()); + Integer maxNumberOfRecords = configuration.getMaxNumberOfRecords(); + if (maxNumberOfRecords != null) { + readTransform = readTransform.withMaxNumberOfRecords(maxNumberOfRecords); } - - if (configuration.getMaxTimeToRun() != null) { - readTransform = readTransform.withMaxTimeToRun(configuration.getMaxTimeToRun()); + Long maxTimeToRun = configuration.getMaxTimeToRun(); + if (maxTimeToRun != null) { + readTransform = readTransform.withMaxTimeToRun(maxTimeToRun); } } @@ -196,13 +196,12 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { @DefaultSchema(AutoValueSchema.class) @AutoValue public abstract static class DebeziumReadSchemaTransformConfiguration { - @Nullable + @SchemaFieldDescription("Maximum number of records to read before stopping.") - public abstract Integer getMaxNumberOfRecords(); + public abstract @Nullable Integer getMaxNumberOfRecords(); - @Nullable @SchemaFieldDescription("Maximum time in milliseconds to run before stopping.") - public abstract Long getMaxTimeToRun(); + public abstract @Nullable Long getMaxTimeToRun(); @SchemaFieldDescription("Username used to connect to the source database.") public abstract String getUsername(); From 315ed95a97dbd8b1024c2fa2d2c6327f64109e59 Mon Sep 17 00:00:00 2001 From: Vitaly Terentyev Date: Tue, 28 Jul 2026 16:30:14 +0400 Subject: [PATCH 7/7] Fix python formatter --- .../apache_beam/yaml/integration_tests.py | 41 +++++++++---------- 1 file changed, 19 insertions(+), 22 deletions(-) diff --git a/sdks/python/apache_beam/yaml/integration_tests.py b/sdks/python/apache_beam/yaml/integration_tests.py index 979bcda44b77..c6d73df76e31 100644 --- a/sdks/python/apache_beam/yaml/integration_tests.py +++ b/sdks/python/apache_beam/yaml/integration_tests.py @@ -21,7 +21,6 @@ import copy import glob import itertools -import json import logging import os import random @@ -39,11 +38,9 @@ import mock import requests -from testcontainers.core.container import DockerContainer from apache_beam.coders import Coder from apache_beam.coders.coder_impl import CoderImpl -from apache_beam.yaml.test_utils.datadog_test_utils import temp_fake_datadog_server class BigEndianIntegerCoderImpl(CoderImpl): @@ -399,11 +396,10 @@ def temp_debezium_postgres_database(): """Provides a temporary PostgreSQL database configured for Debezium CDC.""" container = ( - DockerContainer('quay.io/debezium/example-postgres:latest') - .with_env('POSTGRES_USER', 'debezium') - .with_env('POSTGRES_PASSWORD', 'dbz') - .with_env('POSTGRES_DB', 'inventory') - .with_exposed_ports(5432)) + DockerContainer('quay.io/debezium/example-postgres:latest').with_env( + 'POSTGRES_USER', + 'debezium').with_env('POSTGRES_PASSWORD', 'dbz').with_env( + 'POSTGRES_DB', 'inventory').with_exposed_ports(5432)) try: container.start() @@ -416,29 +412,30 @@ def temp_debezium_postgres_database(): for _ in range(30): try: connection = psycopg2.connect( - host=host, - port=port, - user='debezium', - password='dbz', - dbname='inventory') + host=host, + port=port, + user='debezium', + password='dbz', + dbname='inventory') break except psycopg2.OperationalError: time.sleep(1) if connection is None: - raise RuntimeError('Debezium PostgreSQL container failed to become ready.') + raise RuntimeError( + 'Debezium PostgreSQL container failed to become ready.') try: with connection.cursor() as cursor: cursor.execute( - """ + """ CREATE TABLE IF NOT EXISTS customers ( id INTEGER PRIMARY KEY, name VARCHAR(255) ) """) cursor.execute( - """ + """ INSERT INTO customers (id, name) VALUES (1, 'Alice'), @@ -450,12 +447,12 @@ def temp_debezium_postgres_database(): connection.close() yield { - 'HOST': host, - 'PORT': port, - 'USERNAME': 'debezium', - 'PASSWORD': 'dbz', - 'DATABASE': 'inventory', - 'TABLE': 'public.customers', + 'HOST': host, + 'PORT': port, + 'USERNAME': 'debezium', + 'PASSWORD': 'dbz', + 'DATABASE': 'inventory', + 'TABLE': 'public.customers', } finally: