[incubator-paimon] branch master updated: [flink] validate MySql Config (#1031)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new b3fe12750 [flink] validate MySql Config (#1031)
b3fe12750 is described below

commit b3fe127508e4a9c7321f64bf75b227bc0894318f
Author: zhaoym <zhaoyangm...@deepexi.com>
AuthorDate: Wed Apr 26 19:22:22 2023 +0800

    [flink] validate MySql Config (#1031)
---
 .../flink/action/cdc/mysql/MySqlActionUtils.java   | 31 ++++++++++++++++++++++
 1 file changed, 31 insertions(+)

diff --git 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
index bbbaaeb26..549eb6141 100644
--- 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
+++ 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/mysql/MySqlActionUtils.java
@@ -42,6 +42,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 class MySqlActionUtils {
 
     static Connection getConnection(Configuration mySqlConfig) throws 
Exception {
@@ -117,6 +119,7 @@ class MySqlActionUtils {
     }
 
     static MySqlSource<String> buildMySqlSource(Configuration mySqlConfig) {
+        validateMySqlConfig(mySqlConfig);
         MySqlSourceBuilder<String> sourceBuilder = MySqlSource.builder();
 
         String databaseName = 
mySqlConfig.get(MySqlSourceOptions.DATABASE_NAME);
@@ -202,4 +205,32 @@ class MySqlActionUtils {
                 new JsonDebeziumDeserializationSchema(true, 
customConverterConfigs);
         return 
sourceBuilder.deserializer(schema).includeSchemaChanges(true).build();
     }
+
+    private static void validateMySqlConfig(Configuration mySqlConfig) {
+        checkArgument(
+                mySqlConfig.get(MySqlSourceOptions.HOSTNAME) != null,
+                String.format(
+                        "mysql-conf [%s] must be specified.", 
MySqlSourceOptions.HOSTNAME.key()));
+
+        checkArgument(
+                mySqlConfig.get(MySqlSourceOptions.USERNAME) != null,
+                String.format(
+                        "mysql-conf [%s] must be specified.", 
MySqlSourceOptions.USERNAME.key()));
+
+        checkArgument(
+                mySqlConfig.get(MySqlSourceOptions.PASSWORD) != null,
+                String.format(
+                        "mysql-conf [%s] must be specified.", 
MySqlSourceOptions.PASSWORD.key()));
+
+        checkArgument(
+                mySqlConfig.get(MySqlSourceOptions.DATABASE_NAME) != null,
+                String.format(
+                        "mysql-conf [%s] must be specified.",
+                        MySqlSourceOptions.DATABASE_NAME.key()));
+
+        checkArgument(
+                mySqlConfig.get(MySqlSourceOptions.TABLE_NAME) != null,
+                String.format(
+                        "mysql-conf [%s] must be specified.", 
MySqlSourceOptions.TABLE_NAME.key()));
+    }
 }