optimize code

Author: a49a

…om/dtstack/flink/sql/util/Krb5Utils.java …com/dtstack/flink/sql/util/KrbUtils.javacore/src/main/java/com/dtstack/flink/sql/util/Krb5Utils.java renamed to core/src/main/java/com/dtstack/flink/sql/util/KrbUtils.java core/src/main/java/com/dtstack/flink/sql/util/Krb5Utils.java renamed to core/src/main/java/com/dtstack/flink/sql/util/KrbUtils.java

@@ -28,7 +28,7 @@
  * @author: wuren
  * @create: 2020/09/14
  **/
-public class Krb5Utils {
+public class KrbUtils {
 
     public static final String KRB5_CONF_KEY = "java.security.krb5.conf";
     public static final String HADOOP_AUTH_KEY = "hadoop.security.authentication";

…tstack/flink/sql/util/Krb5UtilsTest.java …dtstack/flink/sql/util/KrbUtilsTest.javacore/src/test/java/com/dtstack/flink/sql/util/Krb5UtilsTest.java renamed to core/src/test/java/com/dtstack/flink/sql/util/KrbUtilsTest.java core/src/test/java/com/dtstack/flink/sql/util/Krb5UtilsTest.java renamed to core/src/test/java/com/dtstack/flink/sql/util/KrbUtilsTest.java

@@ -28,17 +28,16 @@
  * @author: wuren
  * @create: 2020/09/14
  **/
-public class Krb5UtilsTest {
+public class KrbUtilsTest {
     @Test
     public void testGetUgi() throws IOException {
         String principal = "";
         String keytabPath = "";
         String krb5confPath = "";
         try {
-            Krb5Utils.getUgi(principal, keytabPath, krb5confPath);
+            KrbUtils.getUgi(principal, keytabPath, krb5confPath);
         } catch (IllegalArgumentException e) {
             Assert.assertEquals(e.getMessage(), "Can't get Kerberos realm");
         }
-
     }
 }

impala/impala-side/impala-all-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAllReqRow.java

@@ -24,7 +24,7 @@
 import com.dtstack.flink.sql.side.impala.table.ImpalaSideTableInfo;
 import com.dtstack.flink.sql.side.rdb.all.AbstractRdbAllReqRow;
 import com.dtstack.flink.sql.util.JDBCUtils;
-import com.dtstack.flink.sql.util.Krb5Utils;
+import com.dtstack.flink.sql.util.KrbUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -72,7 +72,7 @@ public Connection getConn(String dbUrl, String userName, String password) {
                 String keyTabFilePath = impalaSideTableInfo.getKeyTabFilePath();
                 String krb5FilePath = impalaSideTableInfo.getKrb5FilePath();
                 String principal = impalaSideTableInfo.getPrincipal();
-                UserGroupInformation ugi = Krb5Utils.getUgi(principal, keyTabFilePath, krb5FilePath);
+                UserGroupInformation ugi = KrbUtils.getUgi(principal, keyTabFilePath, krb5FilePath);
                 connection = ugi.doAs(new PrivilegedAction<Connection>() {
                     @Override
                     public Connection run() {

impala/impala-side/impala-async-side/src/main/java/com/dtstack/flink/sql/side/impala/ImpalaAsyncReqRow.java

@@ -23,7 +23,7 @@
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.impala.table.ImpalaSideTableInfo;
 import com.dtstack.flink.sql.side.rdb.async.RdbAsyncReqRow;
-import com.dtstack.flink.sql.util.Krb5Utils;
+import com.dtstack.flink.sql.util.KrbUtils;
 import io.vertx.core.Vertx;
 import io.vertx.core.VertxOptions;
 import io.vertx.core.json.JsonObject;
@@ -72,7 +72,7 @@ public void open(Configuration parameters) throws Exception {
             String keyTabFilePath = impalaSideTableInfo.getKeyTabFilePath();
             String krb5FilePath = impalaSideTableInfo.getKrb5FilePath();
             String principal = impalaSideTableInfo.getPrincipal();
-            ugi = Krb5Utils.getUgi(principal, keyTabFilePath, krb5FilePath);
+            ugi = KrbUtils.getUgi(principal, keyTabFilePath, krb5FilePath);
             openJdbc(parameters);
         } else {
             openJdbc(parameters);

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java

@@ -2,7 +2,7 @@
 
 import com.dtstack.flink.sql.sink.rdb.JDBCOptions;
 import com.dtstack.flink.sql.sink.rdb.format.JDBCUpsertOutputFormat;
-import com.dtstack.flink.sql.util.Krb5Utils;
+import com.dtstack.flink.sql.util.KrbUtils;
 import org.apache.hadoop.security.UserGroupInformation;
 
 import java.io.IOException;
@@ -54,11 +54,15 @@ public ImpalaOutputFormat(
     @Override
     public void open(int taskNumber, int numTasks) throws IOException {
         if (authMech == 1) {
-            UserGroupInformation ugi = Krb5Utils.getUgi(principal, keytabPath, krb5confPath);
+            UserGroupInformation ugi = KrbUtils.getUgi(principal, keytabPath, krb5confPath);
             ugi.doAs(new PrivilegedAction<Object>() {
                 @Override
                 public Object run() {
-                    openJdbc();
+                    try {
+                        openJdbc();
+                    } catch (IOException e) {
+                        e.printStackTrace();
+                    }
                     return null;
                 }
             });

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/JDBCUpsertOutputFormat.java

@@ -113,11 +113,10 @@ public void open(int taskNumber, int numTasks) throws IOException {
         openJdbc();
     }
 
-    public void openJdbc() {
+    public void openJdbc() throws IOException {
         try {
             establishConnection();
             initMetric();
-
             if (StringUtils.equalsIgnoreCase(updateMode, EUpdateMode.APPEND.name()) || keyFields == null || keyFields.length == 0) {
                 String insertSql = dialect.getInsertIntoStatement(schema, tableName, fieldNames, partitionFields);
                 LOG.info("execute insert sql： {}", insertSql);
@@ -128,7 +127,7 @@ public void openJdbc() {
                     getRuntimeContext().getExecutionConfig().isObjectReuseEnabled(), allReplace, this);
             }
             jdbcWriter.open(connection);
-        } catch (SQLException | IOException sqe) {
+        } catch (SQLException sqe) {
             throw new IllegalArgumentException("open() failed.", sqe);
         } catch (ClassNotFoundException cnfe) {
             throw new IllegalArgumentException("JDBC driver class not found.", cnfe);