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
commit 621a78cd4123bf05a525027b531a633347473a6d Author: Vitaly Terentyev <[email protected]> AuthorDate: Tue Jul 21 15:47:32 2026 +0400 Add Beam YAML support for DebeziumIO --- .../DebeziumReadSchemaTransformProvider.java | 76 ++++++++--- .../DebeziumReadSchemaTransformProviderTest.java | 139 +++++++++++++++++++++ sdks/python/apache_beam/yaml/standard_io.yaml | 21 ++++ 3 files changed, 220 insertions(+), 16 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 d1d5103cc49..1a3055e4bf4 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 @@ package org.apache.beam.io.debezium; 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 @@ public class 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<DebeziumReadSchemaTransformConfiguration> configurationClass() { @@ -82,21 +96,15 @@ public class 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<String> 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 class DebeziumReadSchemaTransformProvider 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 class DebeziumReadSchemaTransformProvider 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<String> 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<String> 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 00000000000..07bd91a5f52 --- /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 8c074e0885d..6ee925ba953 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:
