Merge pull request #408 from WTZ468071157/hotfix_1.10_release_407

[hotfix-407][all] fix bugs and add dirty-plugins.

Author: FlechazoW

.gitignore

@@ -16,4 +16,6 @@ bin/nohup.out
 .DS_Store
 bin/sideSql.txt
 *.keytab
-krb5.conf
+krb5.conf
+.gradle
+gradle

.gitlab-ci.yml

@@ -1,10 +1,22 @@
-build:
+stages:
+  - validate
+  - test
+
+test-job:
   stage: test
   script:
     - mvn clean org.jacoco:jacoco-maven-plugin:0.7.8:prepare-agent package -Dmaven.test.failure.ignore=true -q
+  only:
+    - v1.10.0_dev
+  tags:
+    - dt-insight-engine
+
+validate-job:
+  stage: validate
+  script:
     - mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkStreamSQL"  -Dsonar.login=11974c5e9a29625efa09fdc3c3fdc031efb1aab1 -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar
     - sh ci/sonar_notify.sh
   only:
-    - v1.8.0_dev
+    - v1.10.0_dev
   tags:
-    - dt-insight-engine
+    - dt-insight-engine

README.md

@@ -31,7 +31,6 @@ FlinkStreamSQL
 
 ## 目录
 
-[ 1.0 变更记录](docs/changelog.md)  
 [ 1.1 demo](docs/demo.md)  
 [ 1.2 快速开始](docs/quickStart.md)  
 [ 1.3 参数配置](docs/config.md)  

cassandra/pom.xml

@@ -17,12 +17,6 @@
     </modules>
 
     <dependencies>
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>3.8.1</version>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>com.dtstack.flink</groupId>
             <artifactId>sql.core</artifactId>

clickhouse/clickhouse-side/clickhouse-all-side/src/main/java/com/dtstack/flink/sql/side/clickhouse/ClickhouseAllReqRow.java

@@ -18,11 +18,11 @@
 
 package com.dtstack.flink.sql.side.clickhouse;
 
+import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.side.FieldInfo;
 import com.dtstack.flink.sql.side.JoinInfo;
 import com.dtstack.flink.sql.side.AbstractSideTableInfo;
 import com.dtstack.flink.sql.side.rdb.all.AbstractRdbAllReqRow;
-import com.dtstack.flink.sql.util.JDBCUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,7 +45,7 @@ public ClickhouseAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<Fiel
     public Connection getConn(String dbUrl, String userName, String passWord) {
         try {
             Connection connection ;
-            JDBCUtils.forName(CLICKHOUSE_DRIVER, getClass().getClassLoader());
+            ClassLoaderManager.forName(CLICKHOUSE_DRIVER, getClass().getClassLoader());
             // ClickHouseProperties contains all properties
             if (userName == null) {
                 connection = DriverManager.getConnection(dbUrl);

clickhouse/clickhouse-side/clickhouse-side-core/src/main/java/com/dtstack/flink/sql/side/clickhouse/table/ClickhouseSideParser.java

@@ -19,6 +19,7 @@
 
 package com.dtstack.flink.sql.side.clickhouse.table;
 
+import com.dtstack.flink.sql.core.rdb.JdbcCheckKeys;
 import com.dtstack.flink.sql.side.rdb.table.RdbSideParser;
 import com.dtstack.flink.sql.table.AbstractTableInfo;
 import ru.yandex.clickhouse.domain.ClickHouseDataType;
@@ -39,6 +40,7 @@ public class ClickhouseSideParser extends RdbSideParser {
 
     @Override
     public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
+        props.put(JdbcCheckKeys.DRIVER_NAME, "ru.yandex.clickhouse.ClickHouseDriver");
         AbstractTableInfo clickhouseTableInfo = super.getTableInfo(tableName, fieldsInfo, props);
         clickhouseTableInfo.setType(CURR_TYPE);
         return clickhouseTableInfo;

clickhouse/clickhouse-side/pom.xml

@@ -31,6 +31,16 @@
             <artifactId>sql.side.rdb</artifactId>
             <version>${rdb.side.version}</version>
         </dependency>
+
+        <!--    test dependencies   -->
+
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.side.rdb</artifactId>
+            <version>${rdb.side.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

clickhouse/clickhouse-sink/src/main/java/com/dtstack/flink/sql/sink/clickhouse/ClickhouseSink.java

@@ -50,6 +50,7 @@ public JDBCUpsertOutputFormat getOutputFormat() {
                 .setKeyFields(primaryKeys)
                 .setAllReplace(allReplace)
                 .setUpdateMode(updateMode)
+                .setErrorLimit(errorLimit)
                 .build();
     }
 

clickhouse/clickhouse-sink/src/main/java/com/dtstack/flink/sql/sink/clickhouse/table/ClickhouseSinkParser.java

@@ -19,6 +19,7 @@
 
 package com.dtstack.flink.sql.sink.clickhouse.table;
 
+import com.dtstack.flink.sql.core.rdb.JdbcCheckKeys;
 import com.dtstack.flink.sql.sink.rdb.table.RdbSinkParser;
 import com.dtstack.flink.sql.table.AbstractTableInfo;
 import ru.yandex.clickhouse.domain.ClickHouseDataType;
@@ -31,6 +32,7 @@ public class ClickhouseSinkParser extends RdbSinkParser {
 
     @Override
     public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
+        props.put(JdbcCheckKeys.DRIVER_NAME, "ru.yandex.clickhouse.ClickHouseDriver");
         AbstractTableInfo clickhouseTableInfo = super.getTableInfo(tableName, fieldsInfo, props);
         clickhouseTableInfo.setType(CURR_TYPE);
         return clickhouseTableInfo;

console/pom.xml

@@ -16,12 +16,6 @@
     </modules>
 
     <dependencies>
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>3.8.1</version>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>com.dtstack.flink</groupId>
             <artifactId>sql.core</artifactId>

core/pom.xml

@@ -176,6 +176,7 @@
                                         <include name="${project.artifactId}-${project.version}.jar" />
                                     </fileset>
                                 </copy>
+                                <!--suppress UnresolvedMavenProperty -->
                                 <move file="${basedir}/../sqlplugins/${project.artifactId}-${project.version}.jar"
                                       tofile="${basedir}/../sqlplugins/${project.name}-${git.branch}.jar" />
                             </tasks>
@@ -184,7 +185,17 @@
                 </executions>
             </plugin>
 
-
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>test-jar</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 </project>

core/src/main/java/com/dtstack/flink/sql/classloader/ClassLoaderManager.java

@@ -30,6 +30,7 @@
 import java.lang.reflect.Method;
 import java.net.URL;
 import java.net.URLClassLoader;
+import java.sql.DriverManager;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
@@ -45,8 +46,28 @@
 public class ClassLoaderManager {
 
     private static final Logger LOG = LoggerFactory.getLogger(ClassLoaderManager.class);
+    private static final Map<String, DtClassLoader> pluginClassLoader = new ConcurrentHashMap<>();
+    private static final Object LOCK = new Object();
 
-    private static Map<String, DtClassLoader> pluginClassLoader = new ConcurrentHashMap<>();
+    public static void forName(String clazz, ClassLoader classLoader) {
+        synchronized (LOCK) {
+            try {
+                Class.forName(clazz, true, classLoader);
+                DriverManager.setLoginTimeout(10);
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+    public synchronized static void forName(String clazz) {
+        try {
+            Class<?> driverClass = Class.forName(clazz);
+            driverClass.newInstance();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
 
     public static <R> R newInstance(String pluginJarPath, ClassLoaderSupplier<R> supplier) throws Exception {
         ClassLoader classLoader = retrieveClassLoad(pluginJarPath);
@@ -109,11 +130,10 @@ public static List<URL> getClassPath() {
     }
 
 
-
     public static URLClassLoader loadExtraJar(List<URL> jarUrlList, URLClassLoader classLoader)
-            throws  IllegalAccessException, InvocationTargetException {
-        for(URL url : jarUrlList){
-            if(url.toString().endsWith(".jar")){
+            throws IllegalAccessException, InvocationTargetException {
+        for (URL url : jarUrlList) {
+            if (url.toString().endsWith(".jar")) {
                 urlClassLoaderAddUrl(classLoader, url);
             }
         }

core/src/main/java/com/dtstack/flink/sql/exception/ExceptionTrace.java

@@ -1,5 +1,7 @@
 package com.dtstack.flink.sql.exception;
 
+import org.apache.flink.runtime.execution.SuppressRestartsException;
+
 import java.util.Objects;
 
 /**
@@ -18,4 +20,21 @@ public static String traceOriginalCause(Throwable e) {
         }
         return errorMsg;
     }
+
+    /**
+     * 根据异常的种类来判断是否需要强制跳过Flink的重启{@link SuppressRestartsException}
+      * @param e exception
+     * @param errorMsg 需要抛出的异常信息
+     */
+    public static void dealExceptionWithSuppressStart(Exception e, String errorMsg) {
+        if (e instanceof SuppressRestartsException) {
+            throw new SuppressRestartsException(
+                new Throwable(
+                    errorMsg
+                )
+            );
+        } else {
+            throw new RuntimeException(errorMsg);
+        }
+    }
 }