Debezium IO yaml

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 @@
     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<DebeziumReadSchemaTransformConfiguration>
       configurationClass() {
@@ -82,21 +100,15 @@
   @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 +135,9 @@
             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 @@
               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.
-        Schema recordSchema = readTransform.getRecordSchema();
+        Schema recordSchema = withoutOptions(readTransform.getRecordSchema());
         LOG.info(
             "Computed schema for table {} from {}: {}",
             configuration.getTable(),
@@ -172,29 +193,77 @@
     return Collections.singletonList("output");
   }
 
+  @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();
 
+    @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();
     }
 
     @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);

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"));
+  }
+}

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}