[fix] fix some bug

Author: FlechazoW

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

@@ -19,13 +19,15 @@
 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");
     /**
      * 脏数据信息内容
      */
@@ -34,7 +36,7 @@ public class DirtyDataEntity {
     /**
      * 脏数据处理时间
      */
-    private Date processDate;
+    private String processDate;
 
     /**
      * 产生脏数据的原因
@@ -54,11 +56,11 @@ public void setDirtyData(String dirtyData) {
         this.dirtyData = dirtyData;
     }
 
-    public Date getProcessDate() {
+    public String getProcessDate() {
         return processDate;
     }
 
-    public void setProcessDate(Date processDate) {
+    public void setProcessDate(String processDate) {
         this.processDate = processDate;
     }
 
@@ -80,7 +82,7 @@ public void setField(String field) {
 
     public DirtyDataEntity(String dirtyData, Long processDate, String cause, String field) {
         this.dirtyData = dirtyData;
-        this.processDate = new Date(processDate);
+        this.processDate = timeFormat.format(processDate);
         this.cause = cause;
         this.field = field;
     }

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

@@ -123,12 +123,13 @@ private static AbstractDirtyDataConsumer createConsumer(Map<String, String> prop
 
     /**
      * 脏数据收集任务停止，任务停止之前，需要将队列中所有的数据清空
+     * TODO consumer 关闭时仍有数据没有消费到，假如有500条数据，在结束时实际消费数量可能只有493
      */
-    public void close() throws Exception {
-        dirtyDataConsumer.shutdown();
+    public void close() {
         if (checkConsumer()) {
             LOG.info("dirty consumer is closing ...");
             consumer.close();
+            dirtyDataConsumer.shutdownNow();
         }
     }
 

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

@@ -27,7 +27,7 @@ public static void main(String[] args) throws Exception {
             Thread.sleep(100);
             manager.collectDirtyData("testDirtyData" + i,
                     new Exception("testException" + i).getMessage(), "testField");
-            if (i == 500) {
+            if (i == 50) {
                 manager.close();
             }
         }

dirtyDataManager/console/pom.xml

@@ -8,7 +8,7 @@
         <version>1.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
-
+    <name>dirtyConsumer-console</name>
     <artifactId>dirtyConsumer.console</artifactId>
 
     <build>

dirtyDataManager/mysql/pom.xml

@@ -8,7 +8,7 @@
         <version>1.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
-
+    <name>dirtyConsumer-mysql</name>
     <properties>
         <mysql.connector.version>5.1.46</mysql.connector.version>
     </properties>