Merge pull request #483 from data-integrations/PLUGIN-1640-cherrypick

[🍒][PLUGIN-1640] add support for using different ports in cloud sql proxy VM and connection config macro

Author: itsankit-google

aurora-mysql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.10.4</version>
+    <version>1.10.5-SNAPSHOT</version>
   </parent>
 
   <name>Aurora DB MySQL plugin</name>

aurora-postgresql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.10.4</version>
+    <version>1.10.5-SNAPSHOT</version>
   </parent>
 
   <name>Aurora DB PostgreSQL plugin</name>

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

@@ -23,6 +23,8 @@ Properties
 **Connection Name:** The CloudSQL instance to connect to in the format <PROJECT_ID>:\<REGION>:<INSTANCE_NAME>. 
 Can be found in the instance overview page.
 
+**Port:** Port that MySQL is running on.
+
 **CloudSQL Instance Type:** Whether the CloudSQL instance to connect to is private or public. Defaults to 'Public'.
 
 **Username:** User identity for connecting to the specified database.

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

@@ -32,6 +32,8 @@ You also can use the macro function ${conn(connection-name)}.
 **Connection Name:** The CloudSQL instance to connect to in the format <PROJECT_ID>:\<REGION>:<INSTANCE_NAME>. 
 Can be found in the instance overview page.
 
+**Port:** Port that MySQL is running on.
+
 **CloudSQL Instance Type:** Whether the CloudSQL instance to connect to is private or public. Defaults to 'Public'.
 
 **Table Name:** Name of the table to export to. Table must exist prior to running the pipeline.

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

@@ -31,6 +31,8 @@ You also can use the macro function ${conn(connection-name)}.
 **Connection Name:** The CloudSQL instance to connect to in the format <PROJECT_ID>:\<REGION>:<INSTANCE_NAME>.
 Can be found in the instance overview page.
 
+**Port:** Port that MySQL is running on.
+
 **CloudSQL Instance Type:** Whether the CloudSQL instance to connect to is private or public. Defaults to 'Public'.
 
 **Import Query:** The SELECT query to use to import data from the specified table.

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

@@ -18,6 +18,8 @@ Properties
 **Connection Name:** The CloudSQL instance to connect to in the format <PROJECT_ID>:\<REGION>:<INSTANCE_NAME>.
 Can be found in the instance overview page.
 
+**Port:** Port that MySQL is running on.
+
 **Database:** MySQL database name.
 
 **Username:** User identity for connecting to the specified database. Required for databases that need

cloudsql-mysql-plugin/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>database-plugins-parent</artifactId>
     <groupId>io.cdap.plugin</groupId>
-    <version>1.10.4</version>
+    <version>1.10.5-SNAPSHOT</version>
   </parent>
 
   <name>CloudSQL MySQL plugin</name>

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

@@ -18,11 +18,13 @@
 
 import com.google.common.collect.ImmutableMap;
 import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Macro;
 import io.cdap.cdap.api.annotation.Name;
 import io.cdap.cdap.api.annotation.Plugin;
 import io.cdap.cdap.etl.api.FailureCollector;
 import io.cdap.cdap.etl.api.PipelineConfigurer;
 import io.cdap.cdap.etl.api.action.Action;
+import io.cdap.plugin.db.ConnectionConfig;
 import io.cdap.plugin.db.action.AbstractDBAction;
 import io.cdap.plugin.db.action.QueryConfig;
 import io.cdap.plugin.util.CloudSQLUtil;
@@ -48,11 +50,13 @@ public CloudSQLMySQLAction(CloudSQLMySQLActionConfig cloudsqlMysqlActionConfig)
   @Override
   public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
     FailureCollector failureCollector = pipelineConfigurer.getStageConfigurer().getFailureCollector();
-    
-    CloudSQLUtil.checkConnectionName(
-        failureCollector,
-        cloudsqlMysqlActionConfig.instanceType,
-        cloudsqlMysqlActionConfig.connectionName);
+
+    if (cloudsqlMysqlActionConfig.canConnect()) {
+      CloudSQLUtil.checkConnectionName(
+          failureCollector,
+          cloudsqlMysqlActionConfig.instanceType,
+          cloudsqlMysqlActionConfig.connectionName);
+    }
 
     super.configurePipeline(pipelineConfigurer);
   }
@@ -69,10 +73,18 @@ public CloudSQLMySQLActionConfig() {
         "The CloudSQL instance to connect to. For a public instance, the connection string should be in the format "
             + "<PROJECT_ID>:<REGION>:<INSTANCE_NAME> which can be found in the instance overview page. For a private "
             + "instance, enter the internal IP address of the Compute Engine VM cloudsql proxy is running on.")
+    @Macro
     public String connectionName;
 
+    @Name(ConnectionConfig.PORT)
+    @Description("Database port number")
+    @Macro
+    @Nullable
+    private Integer port;
+
     @Name(DATABASE)
     @Description("Database name to connect to")
+    @Macro
     public String database;
 
     @Name(CloudSQLMySQLConstants.CONNECTION_TIMEOUT)
@@ -94,6 +106,7 @@ public String getConnectionString() {
         return String.format(
             CloudSQLMySQLConstants.PRIVATE_CLOUDSQL_MYSQL_CONNECTION_STRING_FORMAT,
             connectionName,
+            getPort(),
             database);
       }
 
@@ -103,10 +116,19 @@ public String getConnectionString() {
           connectionName);
     }
 
+    public int getPort() {
+      return port == null ? 3306 : port;
+    }
+
     @Override
     public Map<String, String> getDBSpecificArguments() {
       return ImmutableMap.of(
           CloudSQLMySQLConstants.CONNECTION_TIMEOUT, String.valueOf(connectionTimeout));
     }
+
+    public boolean canConnect() {
+      return !containsMacro(CloudSQLUtil.CONNECTION_NAME) && !containsMacro(ConnectionConfig.PORT) &&
+          !containsMacro(DATABASE);
+    }
   }
 }

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

@@ -17,6 +17,7 @@
 package io.cdap.plugin.cloudsql.mysql;
 
 import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Macro;
 import io.cdap.cdap.api.annotation.Name;
 import io.cdap.plugin.db.ConnectionConfig;
 import io.cdap.plugin.db.connector.AbstractDBConnectorConfig;
@@ -38,25 +39,35 @@ public class CloudSQLMySQLConnectorConfig extends AbstractDBConnectorConfig {
     "The CloudSQL instance to connect to. For a public instance, the connection string should be in the format "
       + "<PROJECT_ID>:<REGION>:<INSTANCE_NAME> which can be found in the instance overview page. For a private "
       + "instance, enter the internal IP address of the Compute Engine VM cloudsql proxy is running on.")
+  @Macro
   private String connectionName;
 
+  @Name(ConnectionConfig.PORT)
+  @Description("Database port number")
+  @Macro
+  @Nullable
+  private Integer port;
+
   @Name(ConnectionConfig.DATABASE)
   @Description("Database name to connect to")
+  @Macro
   private String database;
 
   @Name(CloudSQLUtil.INSTANCE_TYPE)
   @Description("Whether the CloudSQL instance to connect to is private or public.")
   private String instanceType;
 
   public CloudSQLMySQLConnectorConfig(String user, String password, String jdbcPluginName, String connectionArguments,
-                                      String instanceType, String connectionName, String database) {
+                                      String instanceType, String connectionName, String database,
+                                      @Nullable Integer port) {
     this.user = user;
     this.password = password;
     this.jdbcPluginName = jdbcPluginName;
     this.connectionArguments = connectionArguments;
     this.instanceType = instanceType;
     this.connectionName = connectionName;
     this.database = database;
+    this.port = port;
   }
 
   public String getDatabase() {
@@ -71,12 +82,17 @@ public String getConnectionName() {
     return connectionName;
   }
 
+  public int getPort() {
+    return port == null ? 3306 : port;
+  }
+
   @Override
   public String getConnectionString() {
     if (CloudSQLUtil.PRIVATE_INSTANCE.equalsIgnoreCase(instanceType)) {
       return String.format(
         CloudSQLMySQLConstants.PRIVATE_CLOUDSQL_MYSQL_CONNECTION_STRING_FORMAT,
         connectionName,
+        getPort(),
         database);
     }
 
@@ -93,4 +109,10 @@ public Properties getConnectionArgumentsProperties() {
     properties.put(JDBC_PROPERTY_SOCKET_TIMEOUT_MILLIS, "20000");
     return properties;
   }
+
+  @Override
+  public boolean canConnect() {
+    return super.canConnect() && !containsMacro(CloudSQLUtil.CONNECTION_NAME) &&
+        !containsMacro(ConnectionConfig.PORT) && !containsMacro(ConnectionConfig.DATABASE);
+  }
 }

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

@@ -26,5 +26,5 @@ private CloudSQLMySQLConstants() {
   public static final String CONNECTION_TIMEOUT = "connectionTimeout";
   public static final String PUBLIC_CLOUDSQL_MYSQL_CONNECTION_STRING_FORMAT =
       "jdbc:mysql:///%s?cloudSqlInstance=%s&socketFactory=com.google.cloud.sql.mysql.SocketFactory";
-  public static final String PRIVATE_CLOUDSQL_MYSQL_CONNECTION_STRING_FORMAT = "jdbc:mysql://%s/%s";
+  public static final String PRIVATE_CLOUDSQL_MYSQL_CONNECTION_STRING_FORMAT = "jdbc:mysql://%s:%s/%s";
 }