Merge remote-tracking branch 'origin/1.10_release_4.1.x' into hotfix_1.10_4.1.x_35378

# Conflicts:
#	launcher/src/main/java/com/dtstack/flink/sql/launcher/LauncherMain.java

Author: a49a

.gitignore

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

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>

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

@@ -0,0 +1,100 @@
+/*
+ * 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.consumer;
+
+import com.dtstack.flink.sql.dirtyManager.entity.DirtyDataEntity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * @author tiezhu
+ * Company dtstack
+ * Date 2020/8/27 星期四
+ */
+public abstract class AbstractDirtyDataConsumer implements Runnable, Serializable {
+    protected static final long serialVersionUID = -6058598201315176687L;
+
+    protected static final Logger LOG = LoggerFactory.getLogger(AbstractDirtyDataConsumer.class);
+
+    protected Long errorLimit = 1000L;
+    protected AtomicLong errorCount = new AtomicLong(0L);
+
+    protected AtomicLong count = new AtomicLong(0L);
+
+    public AtomicBoolean isRunning = new AtomicBoolean(true);
+
+    protected LinkedBlockingQueue<DirtyDataEntity> queue;
+
+    /**
+     * 消费队列数据
+     *
+     * @throws Exception throw exception
+     */
+    public abstract void consume() throws Exception;
+
+    /**
+     * 关闭消费者，需要释放资源
+     */
+    public abstract void close();
+
+    /**
+     * 初始化消费者，初始化定时任务
+     *
+     * @param properties 任务参数
+     * @throws Exception throw exception
+     */
+    public abstract void init(Map<String, Object> properties) throws Exception;
+
+    /**
+     * 检验consumer是否正在执行
+     */
+    public boolean isRunning() {
+        return isRunning.get();
+    }
+
+    @Override
+    public void run() {
+        try {
+            while (isRunning.get()) {
+                consume();
+            }
+        } catch (Exception e) {
+            LOG.error("consume dirtyData error", e);
+            if (errorCount.getAndIncrement() > errorLimit) {
+                throw new RuntimeException("The task failed due to the number of dirty data consume failed reached the limit " + errorLimit);
+            }
+        }
+    }
+
+    public AbstractDirtyDataConsumer setQueue(LinkedBlockingQueue<DirtyDataEntity> queue) {
+        this.queue = queue;
+        return this;
+    }
+
+    public void collectDirtyData(DirtyDataEntity dataEntity, long blockingInterval) throws InterruptedException {
+        queue.offer(dataEntity, blockingInterval, TimeUnit.MILLISECONDS);
+    }
+}

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

@@ -0,0 +1,66 @@
+/*
+ * 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.consumer;
+
+import com.dtstack.flink.sql.classloader.ClassLoaderManager;
+import com.dtstack.flink.sql.util.PluginUtil;
+
+import java.io.File;
+import java.lang.reflect.Constructor;
+import java.util.Objects;
+
+import static com.dtstack.flink.sql.util.PluginUtil.upperCaseFirstChar;
+
+/**
+ * @author tiezhu
+ * Company dtstack
+ * Date 2020/12/21 星期一
+ */
+public class DirtyConsumerFactory {
+    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";
+
+    /**
+     * 通过动态方式去加载脏数据插件
+     *
+     * @param dirtyType      脏数据插件类型
+     * @param pluginPath     脏数据插件直地址
+     * @param pluginLoadMode 插件加载方式
+     * @return 脏数据消费者
+     * @throws Exception exception
+     */
+    public static AbstractDirtyDataConsumer getDirtyConsumer(
+            String dirtyType
+            , String pluginPath
+            , String pluginLoadMode) throws Exception {
+        if (Objects.isNull(dirtyType)) {
+            dirtyType = DEFAULT_DIRTY_TYPE;
+        }
+        String consumerType = DIRTY_CONSUMER_PATH + File.separator + dirtyType;
+        String consumerJar = PluginUtil.getJarFileDirPath(consumerType, pluginPath, pluginLoadMode);
+        String className = CLASS_PRE_STR + "." + dirtyType.toLowerCase() + "." + upperCaseFirstChar(dirtyType + CLASS_POST_STR);
+        return ClassLoaderManager.newInstance(consumerJar, cl -> {
+            Class<?> clazz = cl.loadClass(className);
+            Constructor<?> constructor = clazz.getConstructor();
+            return (AbstractDirtyDataConsumer) constructor.newInstance();
+        });
+    }
+}

core/src/main/java/com/dtstack/flink/sql/dirtyManager/entity/DirtyDataEntity.java

@@ -0,0 +1,112 @@
+/*
+ * 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.entity;
+
+import java.sql.Date;
+import java.text.SimpleDateFormat;
+
+/**
+ * @author tiezhu
+ * Company dtstack
+ * Date 2020/8/27 星期四
+ */
+public class DirtyDataEntity {
+    private final SimpleDateFormat timeFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+    /**
+     * 脏数据信息内容
+     */
+    private String dirtyData;
+
+    /**
+     * 脏数据处理时间
+     */
+    private String processDate;
+
+    /**
+     * 产生脏数据的原因
+     */
+    private String cause;
+
+    /**
+     * 产生异常的字段
+     */
+    private String field;
+
+    public String getDirtyData() {
+        return dirtyData;
+    }
+
+    public void setDirtyData(String dirtyData) {
+        this.dirtyData = dirtyData;
+    }
+
+    public String getProcessDate() {
+        return processDate;
+    }
+
+    public void setProcessDate(String processDate) {
+        this.processDate = processDate;
+    }
+
+    public String getCause() {
+        return cause;
+    }
+
+    public void setCause(String cause) {
+        this.cause = cause;
+    }
+
+    public String getField() {
+        return field;
+    }
+
+    public void setField(String field) {
+        this.field = field;
+    }
+
+    public DirtyDataEntity(String dirtyData, Long processDate, String cause, String field) {
+        this.dirtyData = dirtyData;
+        this.processDate = timeFormat.format(processDate);
+        this.cause = cause;
+        this.field = field;
+    }
+
+    public DirtyDataEntity(String dirtyData, Long processDate, String cause) {
+        this.dirtyData = dirtyData;
+        this.processDate = timeFormat.format(processDate);
+        this.cause = cause;
+    }
+
+    @Override
+    public String toString() {
+        return "DirtyDataEntity{" +
+                "dirtyData='" + dirtyData + '\'' +
+                ", processDate=" + processDate +
+                ", cause='" + cause + '\'' +
+                '}';
+    }
+
+    /**
+     * 获取脏数据信息，返回字符数组
+     * @return 脏数据信息字符数组
+     */
+    public String[] get() {
+        return new String[] {dirtyData, String.valueOf(processDate), cause};
+    }
+}