Merge pull request #120 from data-integrations/fix/CDAP-17459

CDAP-17459 Use host name for user as '%' which means the user can connect to MySQL server from any host.

Author: sagarkapare

mysql-delta-plugins/docs/mysql-cdcSource.md

@@ -38,8 +38,8 @@ wait_timeout        = <duration-in-seconds>
 A MySQL user must be defined with all the following permissions on any database which wants to be replicated:
 **SELECT**, **RELOAD**, **SHOW DATABASES**, **REPLICATION SLAVE** and **REPLICATION CLIENT**.
 ```
-mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';
-mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED WITH mysql_native_password BY 'password';
+mysql> CREATE USER 'user'@'%' IDENTIFIED BY 'password';
+mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user'@'%' IDENTIFIED WITH mysql_native_password BY 'password';
 mysql> FLUSH PRIVILEGES;
 ```
 
@@ -64,7 +64,8 @@ slave that is replicating from the server.
 
 **Server Timezone:** Timezone of the MySQL server. This is used when converting dates into timestamps.
 
-**User:** Username to use to connect to the MySQL server.
+**User:** Username to use to connect to the MySQL server. Actual account used by the source while connecting 
+to the MySQL server will be of the form 'user_name'@'%' where user_name is this field.
 
 **Password:** Password to use to connect to the MySQL server.
 

mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlConfig.java

@@ -33,7 +33,8 @@ public class MySqlConfig extends PluginConfig {
   @Description("Port to use to connect to the MySQL server.")
   private int port;
 
-  @Description("Username to use to connect to the MySQL server.")
+  @Description("Username to use to connect to the MySQL server. Actual account used by the source while connecting " +
+    "to the MySQL server will be of the form 'user_name'@'%' where user_name is this field.")
   private String user;
 
   @Macro

mysql-delta-plugins/src/main/java/io/cdap/delta/mysql/MySqlTableAssessor.java

@@ -159,11 +159,12 @@ static ColumnEvaluation evaluateColumn(ColumnDetail detail) throws IllegalArgume
   }
 
   private void checkReplicationPermissions(List<Problem> featureProblems, List<String> permissions) {
-    String query = String.format("SHOW GRANTS FOR '%s'@'%s'", conf.getUser(), conf.getHost());
+    String clientHost = "%";
+    String query = String.format("SHOW GRANTS FOR '%s'@'%s'", conf.getUser(), clientHost);
     Problem permissionUnknownProblem =
       new Problem("Table Replication Permission Unknown",
                   String.format("Unable to check if '%s' permissions were granted for user '%s'@'%s' or not",
-                                String.join(",", permissions), conf.getUser(), conf.getHost()),
+                                String.join(",", permissions), conf.getUser(), clientHost),
                   "Check database connectivity and user permissions",
                   "Change events might fail to be read after the snapshot phase");
     try (Connection connection = DriverManager.getConnection(conf.getJdbcURL(), conf.getConnectionProperties());
@@ -188,9 +189,9 @@ private void checkReplicationPermissions(List<Problem> featureProblems, List<Str
             featureProblems.add(
               new Problem("Table Replication Permission Not Granted",
                           String.format("The '%s' permission is not granted for user '%s'@'%s'", permission,
-                                        conf.getUser(), conf.getHost()),
+                                        conf.getUser(), clientHost),
                           String.format("Grant '%s' permission to user '%s'@'%s'", permission,
-                                        conf.getUser(), conf.getHost()),
+                                        conf.getUser(), clientHost),
                           "Will not be able to read replication events after the initial snapshot completes."));
           }
         }