fix an infinite loop issue

Author: wangzw

gpdbwriter/src/main/java/cn/hashdata/datax/plugin/writer/gpdbwriter/CopyProcessor.java

@@ -45,8 +45,12 @@ public Long call() throws Exception {
 		Thread.currentThread().setName("CopyProcessor");
 		Record record = null;
 
-		while (task.moreRecord() || (record = queueIn.poll(1000L, TimeUnit.MILLISECONDS)) != null) {
-			if (record == null) {
+		while (true) {
+			record = queueIn.poll(1000L, TimeUnit.MILLISECONDS);
+
+			if (record == null && false == task.moreRecord()) {
+				break;
+			} else if (record == null) {
 				continue;
 			}
 

gpdbwriter/src/main/java/cn/hashdata/datax/plugin/writer/gpdbwriter/CopyWorker.java

@@ -21,8 +21,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import com.alibaba.datax.common.element.Column;
-import com.alibaba.datax.common.element.Record;
 import com.alibaba.datax.common.exception.DataXException;
 import com.alibaba.datax.plugin.rdbms.util.DBUtil;
 import com.alibaba.datax.plugin.rdbms.util.DBUtilErrorCode;
@@ -94,8 +92,12 @@ public Long call() throws Exception {
 
 		byte[] data = null;
 		try {
-			while (task.moreData() || (data = queue.poll(1000L, TimeUnit.MILLISECONDS)) != null) {
-				if (data == null) {
+			while (true) {
+				data = queue.poll(1000L, TimeUnit.MILLISECONDS);
+
+				if (data == null && false == task.moreData()) {
+					break;
+				} else if (data == null) {
 					continue;
 				}
 

gpdbwriter/src/main/java/cn/hashdata/datax/plugin/writer/gpdbwriter/CopyWriterTask.java

@@ -76,7 +76,7 @@ private void send(Record record, LinkedBlockingQueue<Record> queue)
 				result.get();
 			}
 
-			Thread.sleep(1000);
+			Thread.sleep(100);
 		}
 	}