Merge pull request #414 from data-integrations/PLUGIN-1640

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

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