console sink add log and serversocketSource modify exception scope

Author: todd5167

console/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/table/TablePrintUtil.java

@@ -1,5 +1,8 @@
 package com.dtstack.flink.sql.sink.console.table;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.ArrayList;
@@ -15,6 +18,7 @@
  * @author xuqianjin
  */
 public class TablePrintUtil {
+    private static final Logger LOG = LoggerFactory.getLogger(TablePrintUtil.class);
     public static final int ALIGN_LEFT = 1;//左对齐
     public static final int ALIGN_RIGHT = 2;//右对齐
     public static final int ALIGN_CENTER = 3;//居中对齐
@@ -222,6 +226,7 @@ public String getTableString() {
      * 直接打印表格
      */
     public void print() {
+        LOG.info("\n"+getTableString());
         System.out.println(getTableString());
     }
 

serversocket/serversocket-source/src/main/java/com/dtstack/flink/sql/source/serversocket/CustomerSocketTextStreamFunction.java

@@ -25,6 +25,8 @@
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.IOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.BufferedReader;
 import java.io.IOException;
@@ -42,11 +44,12 @@
  * @author maqi
  */
 public class CustomerSocketTextStreamFunction implements SourceFunction<Row> {
+	private static final Logger LOG = LoggerFactory.getLogger(CustomerSocketTextStreamFunction.class);
 
 	/**
 	 * Default delay between successive connection attempts.
 	 */
-	private static final int DEFAULT_CONNECTION_RETRY_SLEEP = 500;
+	private static final int DEFAULT_CONNECTION_RETRY_SLEEP = 2000;
 
 	/**
 	 * Default connection timeout when connecting to the server socket (infinite).
@@ -92,32 +95,33 @@ public void run(SourceContext<Row> ctx) throws Exception {
 		long attempt = 0;
 
 		while (isRunning) {
-
-			try (Socket socket = new Socket()) {
+			try {
+				Socket socket = new Socket();
 				currentSocket = socket;
-
 				socket.connect(new InetSocketAddress(tableInfo.getHostname(), tableInfo.getPort()), CONNECTION_TIMEOUT_TIME);
-				try (BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()))) {
-
-					char[] cbuf = new char[8192];
-					int bytesRead;
-					while (isRunning && (bytesRead = reader.read(cbuf)) != -1) {
-						buffer.append(cbuf, 0, bytesRead);
-						int delimPos;
-						String delimiter = tableInfo.getDelimiter();
-						while (buffer.length() >= delimiter.length() && (delimPos = buffer.indexOf(delimiter)) != -1) {
-							String record = buffer.substring(0, delimPos);
-							// truncate trailing carriage return
-							if (delimiter.equals("\n") && record.endsWith("\r")) {
-								record = record.substring(0, record.length() - 1);
-							}
-							ctx.collect(convertToRow(record));
-							buffer.delete(0, delimPos + delimiter.length());
+
+				BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
+				char[] cbuf = new char[8192];
+				int bytesRead;
+				while (isRunning && (bytesRead = reader.read(cbuf)) != -1) {
+					buffer.append(cbuf, 0, bytesRead);
+					int delimPos;
+					String delimiter = tableInfo.getDelimiter();
+					while (buffer.length() >= delimiter.length() && (delimPos = buffer.indexOf(delimiter)) != -1) {
+						String record = buffer.substring(0, delimPos);
+						// truncate trailing carriage return
+						if (delimiter.equals("\n") && record.endsWith("\r")) {
+							record = record.substring(0, record.length() - 1);
 						}
+						ctx.collect(convertToRow(record));
+						buffer.delete(0, delimPos + delimiter.length());
 					}
 				}
+			} catch (Exception e) {
+				LOG.info("Connection server failed, Please check configuration  !!!!!!!!!!!!!!!!");
 			}
 
+
 			// if we dropped out of this loop due to an EOF, sleep and retry
 			if (isRunning) {
 				attempt++;