Merge branch 'hotfix_1.10_4.1.x_34810' into 1.10_release_4.1.x

Author: gituser

core/pom.xml

@@ -108,17 +108,6 @@
             <artifactId>joda-time</artifactId>
             <version>2.5</version>
         </dependency>
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-simple</artifactId>
-            <version>1.7.30</version>
-        </dependency>
-
-        <dependency>
-            <groupId>log4j</groupId>
-            <artifactId>log4j</artifactId>
-            <version>1.2.17</version>
-        </dependency>
     </dependencies>
 
     <build>
@@ -187,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>

core/src/main/java/com/dtstack/flink/sql/dirtyManager/consumer/AbstractDirtyDataConsumer.java

@@ -23,7 +23,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
-import java.util.Properties;
+import java.util.Map;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -66,7 +66,7 @@ public abstract class AbstractDirtyDataConsumer implements Runnable, Serializabl
      * @param properties 任务参数
      * @throws Exception throw exception
      */
-    public abstract void init(Properties properties) throws Exception;
+    public abstract void init(Map<String, Object> properties) throws Exception;
 
     /**
      * 检验consumer是否正在执行
@@ -83,8 +83,7 @@ public void run() {
             }
         } catch (Exception e) {
             LOG.error("consume dirtyData error", e);
-            errorCount.incrementAndGet();
-            if (errorCount.get() > errorLimit) {
+            if (errorCount.getAndIncrement() > errorLimit) {
                 throw new RuntimeException("The task failed due to the number of dirty data consume failed reached the limit " + errorLimit);
             }
         }

core/src/main/java/com/dtstack/flink/sql/dirtyManager/consumer/DirtyConsumerFactory.java

@@ -33,10 +33,10 @@
  * Date 2020/12/21 星期一
  */
 public class DirtyConsumerFactory {
-    private static final String DEFAULT_DIRTY_TYPE = "console";
-    private static final String DIRTY_CONSUMER_PATH = "dirtyData";
-    private static final String CLASS_PRE_STR = "com.dtstack.flink.sql.dirty";
-    private static final String CLASS_POST_STR = "DirtyDataConsumer";
+    public static final String DEFAULT_DIRTY_TYPE = "console";
+    public static final String DIRTY_CONSUMER_PATH = "dirtyData";
+    public static final String CLASS_PRE_STR = "com.dtstack.flink.sql.dirty";
+    public static final String CLASS_POST_STR = "DirtyDataConsumer";
 
     /**
      * 通过动态方式去加载脏数据插件

core/src/main/java/com/dtstack/flink/sql/dirtyManager/manager/DirtyDataManager.java

@@ -27,30 +27,41 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
-import java.util.Properties;
+import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.DEFAULT_BLOCKING_INTERVAL;
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.DEFAULT_ERROR_LIMIT_RATE;
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.DEFAULT_PRINT_LIMIT;
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.DEFAULT_TYPE;
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.DIRTY_BLOCK_STR;
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.DIRTY_LIMIT_RATE_STR;
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.PLUGIN_LOAD_MODE_STR;
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.PLUGIN_PATH_STR;
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.PLUGIN_TYPE_STR;
+import static com.dtstack.flink.sql.dirtyManager.manager.DirtyKeys.PRINT_LIMIT_STR;
+
 /**
  * @author tiezhu
  * Company dtstack
  * Date 2020/8/27 星期四
  */
 public class DirtyDataManager implements Serializable {
+    private static final long serialVersionUID = 1L;
 
     public final static int MAX_POOL_SIZE_LIMIT = 5;
-    private static final long serialVersionUID = 7190970299538893497L;
     private static final Logger LOG = LoggerFactory.getLogger(DirtyDataManager.class);
-    private static final String DIRTY_BLOCK_STR = "blockingInterval";
-    private static final String DIRTY_LIMIT_RATE_STR = "errorLimitRate";
     private final static int MAX_TASK_QUEUE_SIZE = 100;
-    private final static String DEFAULT_ERROR_LIMIT_RATE = "0.8";
-    private final static String DEFAULT_BLOCKING_INTERVAL = "60";
-    public static AbstractDirtyDataConsumer consumer;
 
-    private static ThreadPoolExecutor dirtyDataConsumer;
+    private AbstractDirtyDataConsumer consumer;
+    private transient ThreadPoolExecutor dirtyDataConsumer;
+
+    private static final DirtyDataManager INSTANCE = new DirtyDataManager();
+
     /**
      * 统计manager收集到的脏数据条数
      */
@@ -68,45 +79,65 @@ public class DirtyDataManager implements Serializable {
      */
     private double errorLimitRate;
 
+    private DirtyDataManager() {
+
+    }
+
     /**
      * 通过参数生成manager实例，并同时将consumer实例化
      */
-    public static DirtyDataManager newInstance(Properties properties) {
+    public static DirtyDataManager newInstance(Map<String, Object> properties) {
         try {
-            DirtyDataManager manager = new DirtyDataManager();
-            manager.blockingInterval = Long.parseLong(String.valueOf(properties.getOrDefault(DIRTY_BLOCK_STR, DEFAULT_BLOCKING_INTERVAL)));
-            manager.errorLimitRate = Double.parseDouble(String.valueOf(properties.getOrDefault(DIRTY_LIMIT_RATE_STR, DEFAULT_ERROR_LIMIT_RATE)));
-            consumer = DirtyConsumerFactory.getDirtyConsumer(
-                    properties.getProperty("type")
-                    , properties.getProperty("pluginPath")
-                    , properties.getProperty("pluginLoadMode")
-            );
-            consumer.init(properties);
-            consumer.setQueue(new LinkedBlockingQueue<>());
-            dirtyDataConsumer = new ThreadPoolExecutor(MAX_POOL_SIZE_LIMIT, MAX_POOL_SIZE_LIMIT, 0, TimeUnit.MILLISECONDS,
-                    new LinkedBlockingQueue<>(MAX_TASK_QUEUE_SIZE), new DTThreadFactory("dirtyDataConsumer"), new ThreadPoolExecutor.CallerRunsPolicy());
-            dirtyDataConsumer.execute(consumer);
-            return manager;
+            INSTANCE.setBlockingInterval(Long.parseLong(
+                    String.valueOf(properties.getOrDefault(DIRTY_BLOCK_STR, DEFAULT_BLOCKING_INTERVAL))));
+            INSTANCE.setErrorLimitRate(Double.parseDouble(
+                    String.valueOf(properties.getOrDefault(DIRTY_LIMIT_RATE_STR, DEFAULT_ERROR_LIMIT_RATE))));
+
+            INSTANCE.setConsumer(properties);
+            return INSTANCE;
         } catch (Exception e) {
             throw new RuntimeException("create dirtyManager error!", e);
         }
     }
 
+    private void setConsumer(Map<String, Object> properties) throws Exception {
+        consumer = DirtyConsumerFactory.getDirtyConsumer(
+                String.valueOf(properties.getOrDefault(PLUGIN_TYPE_STR, DEFAULT_TYPE)),
+                String.valueOf(properties.get(PLUGIN_PATH_STR)),
+                String.valueOf(properties.get(PLUGIN_LOAD_MODE_STR))
+        );
+        consumer.init(properties);
+        consumer.setQueue(new LinkedBlockingQueue<>());
+    }
+
+    public void execute() {
+        if (Objects.isNull(dirtyDataConsumer)) {
+            dirtyDataConsumer = new ThreadPoolExecutor(
+                    MAX_POOL_SIZE_LIMIT,
+                    MAX_POOL_SIZE_LIMIT,
+                    0,
+                    TimeUnit.MILLISECONDS,
+                    new LinkedBlockingQueue<>(MAX_TASK_QUEUE_SIZE),
+                    new DTThreadFactory("dirtyDataConsumer", true),
+                    new ThreadPoolExecutor.CallerRunsPolicy());
+            dirtyDataConsumer.execute(consumer);
+        }
+    }
+
     /**
      * 设置脏数据插件默认配置
      *
      * @return console的默认配置
      */
     public static String buildDefaultDirty() {
         JSONObject jsonObject = new JSONObject();
-        jsonObject.put("type", "console");
-        jsonObject.put("printLimit", "1000");
+        jsonObject.put(PLUGIN_TYPE_STR, DEFAULT_TYPE);
+        jsonObject.put(PRINT_LIMIT_STR, DEFAULT_PRINT_LIMIT);
         return jsonObject.toJSONString();
     }
 
     /**
      * 脏数据收集任务停止，任务停止之前，需要将队列中所有的数据清空
-     * TODO consumer 关闭时仍有数据没有消费到，假如有500条数据，在结束时实际消费数量可能只有493
      */
     public void close() {
         if (checkConsumer()) {
@@ -122,13 +153,17 @@ public void close() {
     public void collectDirtyData(String dataInfo, String cause) {
         DirtyDataEntity dirtyDataEntity = new DirtyDataEntity(dataInfo, System.currentTimeMillis(), cause);
         try {
-            consumer.collectDirtyData(dirtyDataEntity, blockingInterval);
             count.incrementAndGet();
-        } catch (Exception ignored) {
+            consumer.collectDirtyData(dirtyDataEntity, blockingInterval);
+        } catch (Exception e) {
             LOG.warn("dirty Data insert error ... Failed number: " + errorCount.incrementAndGet());
-            LOG.warn("error dirty data:" + dirtyDataEntity.toString());
+            LOG.warn("error cause: " + e.getMessage());
+            LOG.warn("error dirty data:" + dirtyDataEntity.getDirtyData());
             if (errorCount.get() > Math.ceil(count.longValue() * errorLimitRate)) {
-                throw new RuntimeException(String.format("The number of failed number 【%s】 reaches the limit, manager fails", errorCount.get()));
+                // close consumer and manager
+                close();
+                throw new RuntimeException(
+                        String.format("The number of failed number 【%s】 reaches the limit, manager fails", errorCount.get()));
             }
         }
     }
@@ -139,4 +174,28 @@ public void collectDirtyData(String dataInfo, String cause) {
     public boolean checkConsumer() {
         return consumer.isRunning();
     }
+
+    public AtomicLong getCount() {
+        return count;
+    }
+
+    public AtomicLong getErrorCount() {
+        return errorCount;
+    }
+
+    public long getBlockingInterval() {
+        return blockingInterval;
+    }
+
+    public void setBlockingInterval(long blockingInterval) {
+        this.blockingInterval = blockingInterval;
+    }
+
+    public double getErrorLimitRate() {
+        return errorLimitRate;
+    }
+
+    public void setErrorLimitRate(double errorLimitRate) {
+        this.errorLimitRate = errorLimitRate;
+    }
 }

core/src/main/java/com/dtstack/flink/sql/dirtyManager/manager/DirtyKeys.java

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.dirtyManager.manager;
+
+/**
+ * Date: 2021/1/6
+ *
+ * @author tiezhu
+ * Company dtstack
+ */
+public class DirtyKeys {
+    public final static String DEFAULT_TYPE = "console";
+    public final static String DEFAULT_BLOCKING_INTERVAL = "60";
+    public final static String DEFAULT_ERROR_LIMIT_RATE = "0.8";
+    public final static String DEFAULT_PRINT_LIMIT = "1000";
+
+    public final static String DIRTY_BLOCK_STR = "blockingInterval";
+    public final static String DIRTY_LIMIT_RATE_STR = "errorLimitRate";
+    public final static String PLUGIN_TYPE_STR = "type";
+    public final static String PLUGIN_PATH_STR = "pluginPath";
+    public final static String PLUGIN_LOAD_MODE_STR = "pluginLoadMode";
+
+    public final static String PRINT_LIMIT_STR = "printLimit";
+
+    public final static String ENV_CLASSLOADER = "envClassLoader";
+    public final static String TABLE_NAME = "tableName";
+}