Merge pull request #206 from data-integrations/enable-connection-for-source/sink

Enable connection for source/sink

Author: yaojiefeng

cloudsql-mysql-plugin/docs/CloudSQLMySQL-batchsink.md

@@ -19,6 +19,11 @@ Properties
 ----------
 **Reference Name:** Name used to uniquely identify this sink for lineage, annotating metadata, etc.
 
+**Use Connection** Whether to use a connection. If a connection is used, you do not need to provide the credentials.
+
+**Connection** Name of the connection to use. Project and service account information will be provided by the connection.
+You also can use the macro function ${conn(connection-name)}.
+
 **Driver Name:** Name of the JDBC driver to use.
 
 **Database:** MySQL database name.

cloudsql-mysql-plugin/docs/CloudSQLMySQL-batchsource.md

@@ -18,6 +18,11 @@ Properties
 ----------
 **Reference Name:** Name used to uniquely identify this source for lineage, annotating metadata, etc.
 
+**Use Connection** Whether to use a connection. If a connection is used, you do not need to provide the credentials.
+
+**Connection** Name of the connection to use. Project and service account information will be provided by the connection.
+You also can use the macro function ${conn(connection-name)}.
+
 **Driver Name:** Name of the JDBC driver to use.
 
 **Database:** MySQL database name.

cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnector.java

@@ -85,7 +85,7 @@ protected String getTableQuery(String database, String schema, String table, int
   protected void setConnectorSpec(ConnectorSpecRequest request, DBConnectorPath path,
                                   ConnectorSpec.Builder builder) {
     Map<String, String> properties = new HashMap<>();
-    setConnectionProperties(properties);
+    setConnectionProperties(properties, request);
     builder
       .addRelatedPlugin(new PluginSpec(CloudSQLMySQLConstants.PLUGIN_NAME, BatchSource.PLUGIN_TYPE, properties))
       .addRelatedPlugin(new PluginSpec(CloudSQLMySQLConstants.PLUGIN_NAME, BatchSink.PLUGIN_TYPE, properties));
@@ -103,14 +103,4 @@ protected void setConnectorSpec(ConnectorSpecRequest request, DBConnectorPath pa
     properties.put(Constants.Reference.REFERENCE_NAME, ReferenceNames.cleanseReferenceName(table));
     properties.put(CloudSQLMySQLSink.CloudSQLMySQLSinkConfig.TABLE_NAME, table);
   }
-
-  @Override
-  protected void setConnectionProperties(Map<String, String> properties) {
-    Map<String, String> rawProperties = config.getRawProperties().getProperties();
-    properties.put(ConnectionConfig.JDBC_PLUGIN_NAME, rawProperties.get(ConnectionConfig.JDBC_PLUGIN_NAME));
-    properties.put(ConnectionConfig.USER, rawProperties.get(ConnectionConfig.USER));
-    properties.put(ConnectionConfig.PASSWORD, rawProperties.get(ConnectionConfig.PASSWORD));
-    properties.put(CloudSQLMySQLConstants.CONNECTION_NAME, rawProperties.get(CloudSQLMySQLConstants.CONNECTION_NAME));
-    properties.put(CloudSQLMySQLConstants.INSTANCE_TYPE, rawProperties.get(CloudSQLMySQLConstants.INSTANCE_TYPE));
-  }
 }

cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLSink.java

@@ -106,6 +106,12 @@ public Map<String, String> getDBSpecificArguments() {
           CloudSQLMySQLConstants.CONNECTION_TIMEOUT, String.valueOf(connectionTimeout));
     }
 
+    @Override
+    public void validate(FailureCollector collector) {
+      ConfigUtil.validateConnection(this, useConnection, connection, collector);
+      super.validate(collector);
+    }
+
     @Override
     @Nullable
     protected CloudSQLMySQLConnectorConfig getConnection() {

cloudsql-mysql-plugin/src/main/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLSource.java

@@ -112,6 +112,12 @@ protected Map<String, String> getDBSpecificArguments() {
       return arguments;
     }
 
+    @Override
+    public void validate(FailureCollector collector) {
+      ConfigUtil.validateConnection(this, useConnection, connection, collector);
+      super.validate(collector);
+    }
+
     @Override
     protected CloudSQLMySQLConnectorConfig getConnection() {
       return connection;

cloudsql-mysql-plugin/widgets/CloudSQLMySQL-batchsink.json

@@ -8,9 +8,18 @@
       "label": "Connection",
       "properties": [
         {
-          "widget-type": "hidden",
+          "widget-type": "toggle",
+          "label": "Use connection",
           "name": "useConnection",
           "widget-attributes": {
+            "on": {
+              "value": "true",
+              "label": "YES"
+            },
+            "off": {
+              "value": "false",
+              "label": "NO"
+            },
             "default": "false"
           }
         },

cloudsql-mysql-plugin/widgets/CloudSQLMySQL-batchsource.json

@@ -8,9 +8,18 @@
       "label": "Connection",
       "properties": [
         {
-          "widget-type": "hidden",
+          "widget-type": "toggle",
+          "label": "Use connection",
           "name": "useConnection",
           "widget-attributes": {
+            "on": {
+              "value": "true",
+              "label": "YES"
+            },
+            "off": {
+              "value": "false",
+              "label": "NO"
+            },
             "default": "false"
           }
         },

cloudsql-postgresql-plugin/docs/CloudSQLPostgreSQL-batchsink.md

@@ -19,6 +19,11 @@ Properties
 ----------
 **Reference Name:** Name used to uniquely identify this sink for lineage, annotating metadata, etc.
 
+**Use Connection** Whether to use a connection. If a connection is used, you do not need to provide the credentials.
+
+**Connection** Name of the connection to use. Project and service account information will be provided by the connection.
+You also can use the macro function ${conn(connection-name)}.
+
 **Driver Name:** Name of the JDBC driver to use.
 
 **Database:** CloudSQL PostgreSQL database name.

cloudsql-postgresql-plugin/docs/CloudSQLPostgreSQL-batchsource.md

@@ -18,6 +18,11 @@ Properties
 ----------
 **Reference Name:** Name used to uniquely identify this source for lineage, annotating metadata, etc.
 
+**Use Connection** Whether to use a connection. If a connection is used, you do not need to provide the credentials.
+
+**Connection** Name of the connection to use. Project and service account information will be provided by the connection.
+You also can use the macro function ${conn(connection-name)}.
+
 **Driver Name:** Name of the JDBC driver to use.
 
 **Database:** CloudSQL PostgreSQL database name.

cloudsql-postgresql-plugin/src/main/java/io/cdap/plugin/cloudsql/postgres/CloudSQLPostgreSQLConnector.java

@@ -91,7 +91,7 @@ protected String getTableQuery(String database, String schema, String table, int
   protected void setConnectorSpec(ConnectorSpecRequest request, DBConnectorPath path,
                                   ConnectorSpec.Builder builder) {
     Map<String, String> properties = new HashMap<>();
-    setConnectionProperties(properties);
+    setConnectionProperties(properties, request);
     builder
       .addRelatedPlugin(new PluginSpec(CloudSQLPostgreSQLConstants.PLUGIN_NAME, BatchSource.PLUGIN_TYPE, properties))
       .addRelatedPlugin(new PluginSpec(CloudSQLPostgreSQLConstants.PLUGIN_NAME, BatchSink.PLUGIN_TYPE, properties));
@@ -108,16 +108,4 @@ protected void setConnectorSpec(ConnectorSpecRequest request, DBConnectorPath pa
     properties.put(Constants.Reference.REFERENCE_NAME, ReferenceNames.cleanseReferenceName(table));
     properties.put(CloudSQLPostgreSQLSink.CloudSQLPostgreSQLSinkConfig.TABLE_NAME, table);
   }
-
-  @Override
-  protected void setConnectionProperties(Map<String, String> properties) {
-    Map<String, String> rawProperties = config.getRawProperties().getProperties();
-    properties.put(ConnectionConfig.JDBC_PLUGIN_NAME, rawProperties.get(ConnectionConfig.JDBC_PLUGIN_NAME));
-    properties.put(ConnectionConfig.USER, rawProperties.get(ConnectionConfig.USER));
-    properties.put(ConnectionConfig.PASSWORD, rawProperties.get(ConnectionConfig.PASSWORD));
-    properties.put(CloudSQLPostgreSQLConstants.CONNECTION_NAME,
-                   rawProperties.get(CloudSQLPostgreSQLConstants.CONNECTION_NAME));
-    properties.put(CloudSQLPostgreSQLConstants.INSTANCE_TYPE,
-                   rawProperties.get(CloudSQLPostgreSQLConstants.INSTANCE_TYPE));
-  }
 }