bugfix

src/main/java/com/winhc/data/task/SinkTask.java

@@ -62,7 +62,9 @@ public class SinkTask extends Thread {
                 }
                 batch.add(sql);
                 sum.addAndGet(1);
-                batch();
+                if (sum.get() % batchNum == 0) {
+                    batch();
+                }
             }
             //last batch
             batch();
@@ -72,14 +74,12 @@ public class SinkTask extends Thread {
     }
 
     private void batch() throws SQLException {
-        if (sum.get() % batchNum == 0) {
-            if (!batch.isEmpty()) {
-                for (String r : batch) {
-                    stmt.addBatch(r);
-                }
-                stmt.executeBatch();
-                batch.clear();
+        if (!batch.isEmpty()) {
+            for (String r : batch) {
+                stmt.addBatch(r);
             }
+            stmt.executeBatch();
+            batch.clear();
             logger.info(String.format("%s %s-%s-total process %s !", ds, Thread.currentThread().getName(), Thread.currentThread().getId(), sum.get()));
         }
     }