This is an automated email from the ASF dual-hosted git repository.
Amar3tto pushed a commit to branch debezium-io-yaml
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/debezium-io-yaml by this push:
new 01c668b1455 Add Debezium YAML integration test
01c668b1455 is described below
commit 01c668b1455c203688e29a4d8017a2e3181a1ef1
Author: Vitaly Terentyev <[email protected]>
AuthorDate: Thu Jul 23 16:02:55 2026 +0400
Add Debezium YAML integration test
---
.../yaml/extended_tests/databases/debezium.yaml | 48 +++++++++++++++
sdks/python/apache_beam/yaml/integration_tests.py | 70 +++++++++++++++++++++-
sdks/python/build.gradle | 4 +-
3 files changed, 120 insertions(+), 2 deletions(-)
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 00000000000..c4f7e3dbdee
--- /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 4fca20ade96..979bcda44b7 100644
--- a/sdks/python/apache_beam/yaml/integration_tests.py
+++ b/sdks/python/apache_beam/yaml/integration_tests.py
@@ -48,7 +48,7 @@ from apache_beam.yaml.test_utils.datadog_test_utils import
temp_fake_datadog_ser
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
[email protected]
+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 9e2fe232c42..b50660978b2 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'
