PLUGIN-1640: add support for using different ports in cloud sql proxy VM

Author: itsankit-google

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/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;
@@ -71,6 +73,12 @@ public CloudSQLMySQLActionConfig() {
             + "instance, enter the internal IP address of the Compute Engine VM cloudsql proxy is running on.")
     public String connectionName;
 
+    @Name(ConnectionConfig.PORT)
+    @Description("Database port number")
+    @Macro
+    @Nullable
+    private Integer port;
+
     @Name(DATABASE)
     @Description("Database name to connect to")
     public String database;
@@ -94,6 +102,7 @@ public String getConnectionString() {
         return String.format(
             CloudSQLMySQLConstants.PRIVATE_CLOUDSQL_MYSQL_CONNECTION_STRING_FORMAT,
             connectionName,
+            getPort(),
             database);
       }
 
@@ -103,6 +112,10 @@ public String getConnectionString() {
           connectionName);
     }
 
+    public Integer getPort() {
+      return port == null ? 3306 : port;
+    }
+
     @Override
     public Map<String, String> getDBSpecificArguments() {
       return ImmutableMap.of(

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;
@@ -40,6 +41,12 @@ public class CloudSQLMySQLConnectorConfig extends AbstractDBConnectorConfig {
       + "instance, enter the internal IP address of the Compute Engine VM cloudsql proxy is running on.")
   private String connectionName;
 
+  @Name(ConnectionConfig.PORT)
+  @Description("Database port number")
+  @Macro
+  @Nullable
+  private Integer port;
+
   @Name(ConnectionConfig.DATABASE)
   @Description("Database name to connect to")
   private String database;
@@ -49,14 +56,16 @@ public class CloudSQLMySQLConnectorConfig extends AbstractDBConnectorConfig {
   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 +80,17 @@ public String getConnectionName() {
     return connectionName;
   }
 
+  public Integer 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);
     }
 

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

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

@@ -91,7 +91,8 @@ protected LineageRecorder getLineageRecorder(BatchSinkContext context) {
       host = connectionParams[2];
       location = connectionParams[1];
     }
-    String fqn = DBUtils.constructFQN("mysql", host, 3306,
+    String fqn = DBUtils.constructFQN("mysql", host,
+                                      cloudsqlMysqlSinkConfig.getConnection().getPort(),
                                       cloudsqlMysqlSinkConfig.getConnection().getDatabase(),
                                       cloudsqlMysqlSinkConfig.getReferenceName());
     Asset.Builder assetBuilder = Asset.builder(cloudsqlMysqlSinkConfig.getReferenceName()).setFqn(fqn);

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

@@ -86,6 +86,7 @@ protected String createConnectionString() {
       return String.format(
           CloudSQLMySQLConstants.PRIVATE_CLOUDSQL_MYSQL_CONNECTION_STRING_FORMAT,
           cloudsqlMysqlSourceConfig.connection.getConnectionName(),
+          cloudsqlMysqlSourceConfig.connection.getPort(),
           cloudsqlMysqlSourceConfig.connection.getDatabase());
     }
 
@@ -108,7 +109,8 @@ protected LineageRecorder getLineageRecorder(BatchSourceContext context) {
       host = connectionParams[2];
       location = connectionParams[1];
     }
-    String fqn = DBUtils.constructFQN("mysql", host, 3306,
+    String fqn = DBUtils.constructFQN("mysql", host,
+                                      cloudsqlMysqlSourceConfig.getConnection().getPort(),
                                       cloudsqlMysqlSourceConfig.getConnection().getDatabase(),
                                       cloudsqlMysqlSourceConfig.getReferenceName());
     Asset.Builder assetBuilder = Asset.builder(cloudsqlMysqlSourceConfig.getReferenceName()).setFqn(fqn);

cloudsql-mysql-plugin/src/test/java/io/cdap/plugin/cloudsql/mysql/CloudSQLMySQLConnectorTest.java

@@ -54,7 +54,7 @@ public void test() throws IOException, ClassNotFoundException, InstantiationExce
     test(
       new CloudSQLMySQLConnector(
         new CloudSQLMySQLConnectorConfig(username, password, JDBC_PLUGIN_NAME, connectionArguments, instanceType,
-                                         connectionName, database)
+                                         connectionName, database, null)
       ),
       JDBC_DRIVER_CLASS_NAME,
       CloudSQLMySQLConstants.PLUGIN_NAME