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..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 @@ -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,23 @@ 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); + } 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 +100,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 +135,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); } } @@ -138,10 +150,19 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { readTransform .withMaxNumberOfRecords(testLimitRecords) .withMaxTimeToRun(testLimitMilliseconds); + } else { + Integer maxNumberOfRecords = configuration.getMaxNumberOfRecords(); + if (maxNumberOfRecords != null) { + readTransform = readTransform.withMaxNumberOfRecords(maxNumberOfRecords); + } + Long maxTimeToRun = configuration.getMaxTimeToRun(); + if (maxTimeToRun != null) { + readTransform = readTransform.withMaxTimeToRun(maxTimeToRun); + } } // 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(), @@ -172,22 +193,65 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { return Collections.singletonList("output"); } + @DefaultSchema(AutoValueSchema.class) @AutoValue public abstract static class DebeziumReadSchemaTransformConfiguration { + + @SchemaFieldDescription("Maximum number of records to read before stopping.") + public abstract @Nullable Integer getMaxNumberOfRecords(); + + @SchemaFieldDescription("Maximum time in milliseconds to run before stopping.") + public abstract @Nullable Long getMaxTimeToRun(); + + @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(); @@ -195,6 +259,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); 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/extended_tests/databases/debezium.yaml b/sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml new file mode 100644 index 000000000000..b8ada54532f4 --- /dev/null +++ b/sdks/python/apache_beam/yaml/extended_tests/databases/debezium.yaml @@ -0,0 +1,50 @@ +# +# 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]}" + max_number_of_records: 2 + max_time_to_run: 600000 + 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..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,16 +38,14 @@ 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): """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 +391,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/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml index 8c074e0885dd..796429b5bdfc 100644 --- a/sdks/python/apache_beam/yaml/standard_io.yaml +++ b/sdks/python/apache_beam/yaml/standard_io.yaml @@ -182,6 +182,29 @@ 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' + max_number_of_records: 'max_number_of_records' + max_time_to_run: 'max_time_to_run' + 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: 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'