Minor changes on Managed JDBCIO (#36339)

Author: shunping

sdks/java/expansion-service/container/Dockerfile

@@ -28,7 +28,6 @@ WORKDIR /opt/apache/beam
 COPY target/avro.jar jars/
 COPY target/beam-sdks-java-io-expansion-service.jar jars/
 COPY target/beam-sdks-java-io-google-cloud-platform-expansion-service.jar jars/
-COPY target/beam-sdks-java-extensions-schemaio-expansion-service.jar jars/
 
 # Copy licenses
 COPY target/LICENSE /opt/apache/beam/

sdks/java/extensions/schemaio-expansion-service/build.gradle

@@ -76,3 +76,9 @@ task runExpansionService (type: JavaExec) {
     classpath = sourceSets.test.runtimeClasspath
     args = [project.findProperty("constructionService.port") ?: "8097"]
 }
+
+shadowJar {
+    manifest {
+        attributes(["Multi-Release": true])
+    }
+}

sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/providers/ReadFromPostgresSchemaTransformProvider.java

@@ -59,14 +59,18 @@ protected String jdbcType() {
       JdbcReadSchemaTransformConfiguration configuration) {
     String jdbcType = configuration.getJdbcType();
     if (jdbcType != null && !jdbcType.isEmpty() && !jdbcType.equals(jdbcType())) {
-      throw new IllegalArgumentException(
-          String.format("Wrong JDBC type. Expected '%s' but got '%s'", jdbcType(), jdbcType));
+      LOG.warn(
+          "Wrong JDBC type. Expected '{}' but got '{}'. Overriding with '{}'.",
+          jdbcType(),
+          jdbcType,
+          jdbcType());
+      configuration = configuration.toBuilder().setJdbcType(jdbcType()).build();
     }
 
     List<@org.checkerframework.checker.nullness.qual.Nullable String> connectionInitSql =
         configuration.getConnectionInitSql();
     if (connectionInitSql != null && !connectionInitSql.isEmpty()) {
-      LOG.warn("Postgres does not support connectionInitSql, ignoring.");
+      throw new IllegalArgumentException("Postgres does not support connectionInitSql.");
     }
 
     Boolean disableAutoCommit = configuration.getDisableAutoCommit();

sdks/java/io/jdbc/src/main/java/org/apache/beam/sdk/io/jdbc/providers/WriteToPostgresSchemaTransformProvider.java

@@ -59,14 +59,18 @@ protected String jdbcType() {
       JdbcWriteSchemaTransformConfiguration configuration) {
     String jdbcType = configuration.getJdbcType();
     if (jdbcType != null && !jdbcType.isEmpty() && !jdbcType.equals(jdbcType())) {
-      throw new IllegalArgumentException(
-          String.format("Wrong JDBC type. Expected '%s' but got '%s'", jdbcType(), jdbcType));
+      LOG.warn(
+          "Wrong JDBC type. Expected '{}' but got '{}'. Overriding with '{}'.",
+          jdbcType(),
+          jdbcType,
+          jdbcType());
+      configuration = configuration.toBuilder().setJdbcType(jdbcType()).build();
     }
 
     List<@org.checkerframework.checker.nullness.qual.Nullable String> connectionInitSql =
         configuration.getConnectionInitSql();
     if (connectionInitSql != null && !connectionInitSql.isEmpty()) {
-      LOG.warn("Postgres does not support connectionInitSql, ignoring.");
+      throw new IllegalArgumentException("Postgres does not support connectionInitSql.");
     }
 
     // Override "connectionInitSql" for postgres

sdks/python/apache_beam/transforms/external.py

@@ -85,6 +85,8 @@
     ManagedTransforms.Urns.POSTGRES_WRITE.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,  # pylint: disable=line-too-long
     ManagedTransforms.Urns.MYSQL_READ.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,
     ManagedTransforms.Urns.MYSQL_WRITE.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,
+    ManagedTransforms.Urns.SQL_SERVER_READ.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,  # pylint: disable=line-too-long
+    ManagedTransforms.Urns.SQL_SERVER_WRITE.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,  # pylint: disable=line-too-long
 }