refactor logic to check copy error. Report copy error first

Author: wangzw

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

@@ -55,8 +55,6 @@ public Long call() throws Exception {
 				try {
 					CopyManager mgr = new CopyManager((BaseConnection) connection);
 					return mgr.copyIn(sql, pipeIn);
-				} catch (PSQLException e) {
-					throw new IOException("无法向目标表写入数据: " + e.getMessage());
 				} finally {
 					try {
 						pipeIn.close();
@@ -72,18 +70,6 @@ public Long call() throws Exception {
 		copyBackendThread.start();
 	}
 
-	public void write(byte[] record) throws InterruptedException {
-		queue.put(record);
-
-		if (copyResult.isDone()) {
-			try {
-				copyResult.get();
-			} catch (ExecutionException e) {
-				throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, e);
-			}
-		}
-	}
-
 	@Override
 	public Long call() throws Exception {
 		Thread.currentThread().setName("CopyWorker");
@@ -104,8 +90,6 @@ public Long call() throws Exception {
 
 			pipeOut.flush();
 			pipeOut.close();
-			Long count = copyResult.get();
-			return count;
 		} catch (Exception e) {
 			try {
 				((BaseConnection) connection).cancelQuery();
@@ -118,6 +102,16 @@ public Long call() throws Exception {
 			} catch (SecurityException ignore) {
 			}
 
+			try {
+				copyResult.get();
+			} catch (ExecutionException exec) {
+				if (exec.getCause() instanceof PSQLException) {
+					throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, exec.getCause());
+				}
+				// ignore others
+			} catch (Exception ignore) {
+			}
+
 			throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, e);
 		} finally {
 			try {
@@ -134,6 +128,13 @@ public Long call() throws Exception {
 
 			DBUtil.closeDBResources(null, null, connection);
 		}
+
+		try {
+			Long count = copyResult.get();
+			return count;
+		} catch (Exception e) {
+			throw DataXException.asDataXException(DBUtilErrorCode.WRITE_DATA_ERROR, e);
+		}
 	}
 
 	private void changeCsvSizelimit(Connection conn) {