Merge branch 'hotfix_1.8_4.0.x_30332' into 1.8_release_4.0.x

Author: gituser

flinkx-metadata-oracle/flinkx-metadata-oracle-reader/src/main/java/com/dtstack/flinkx/metadataoracle/inputformat/MetadataoracleInputFormat.java

@@ -21,6 +21,7 @@
 import com.dtstack.flinkx.constants.ConstantValue;
 import com.dtstack.flinkx.metadata.inputformat.BaseMetadataInputFormat;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -105,7 +106,7 @@ protected Map<String, Object> queryMetaData(String tableName) throws SQLExceptio
 
     Map<String, Map<String, String> > queryTableProperties() throws SQLException {
         Map<String, Map<String, String>> tablePropertiesMap = new HashMap<>(16);
-        if(allTable==null){
+        if(StringUtils.isBlank(allTable)){
             sql = String.format(SQL_QUERY_TABLE_PROPERTIES_TOTAL, quote(currentDb.get()));
         }else {
             sql = String.format(SQL_QUERY_TABLE_PROPERTIES, quote(currentDb.get()), allTable);
@@ -126,7 +127,7 @@ Map<String, Map<String, String> > queryTableProperties() throws SQLException {
 
     Map<String, List<Map<String, String>>> queryIndexList() throws SQLException {
         Map<String, List<Map<String, String>>> indexListMap = new HashMap<>(16);
-        if(allTable==null){
+        if(StringUtils.isBlank(allTable)){
             sql = String.format(SQL_QUERY_INDEX_TOTAL, quote(currentDb.get()));
         }else {
             sql = String.format(SQL_QUERY_INDEX, quote(currentDb.get()), allTable);
@@ -152,7 +153,7 @@ Map<String, List<Map<String, String>>> queryIndexList() throws SQLException {
 
     Map<String, List<Map<String, Object>>> queryColumnList() throws SQLException {
         Map<String, List<Map<String, Object>>> columnListMap = new HashMap<>(16);
-        if(allTable==null){
+        if(StringUtils.isBlank(allTable)){
             sql = String.format(SQL_QUERY_COLUMN_TOTAL, quote(currentDb.get()));
         }else {
             sql = String.format(SQL_QUERY_COLUMN, quote(currentDb.get()), allTable);

flinkx-metadata/flinkx-metadata-reader/src/main/java/com/dtstack/flinkx/metadata/inputformat/BaseMetadataInputFormat.java

@@ -82,8 +82,9 @@ protected void openInternal(InputSplit inputSplit) throws IOException {
             tableIterator.set(tableList.iterator());
             init();
         } catch (SQLException | ClassNotFoundException e) {
-            LOG.error("获取table列表异常, dbUrl = {}, username = {}, inputSplit = {}, e = {}", dbUrl, username, inputSplit, ExceptionUtil.getErrorMessage(e));
-            throw new IOException("获取table列表异常", e);
+            String message = String.format("获取table列表异常, dbUrl = %s, username = %s, inputSplit = %s, e = %s", dbUrl, username, inputSplit, ExceptionUtil.getErrorMessage(e));
+            LOG.error(message);
+            throw new IOException(message, e);
         }
     }
 

flinkx-restapi/flinkx-restapi-writer/src/main/java/com/dtstack/flinkx/restapi/outputformat/RestapiOutputFormat.java

@@ -22,6 +22,8 @@
 import com.dtstack.flinkx.restapi.common.HttpUtil;
 import com.dtstack.flinkx.util.ExceptionUtil;
 import com.google.common.collect.Maps;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
 import org.apache.flink.types.Row;
 import org.apache.http.HttpStatus;
 import org.apache.http.client.config.RequestConfig;
@@ -57,9 +59,12 @@ public class RestapiOutputFormat extends BaseRichOutputFormat {
 
     protected static final int DEFAULT_TIME_OUT = 300000;
 
+    protected Gson gson;
+
     @Override
     protected void openInternal(int taskNumber, int numTasks) throws IOException {
         params.put("threadId", UUID.randomUUID().toString().substring(0, 8));
+        gson = new GsonBuilder().serializeNulls().create();
     }
 
     @Override
@@ -80,7 +85,7 @@ protected void writeSingleRecordInternal(Row row) throws WriteRecordException {
             }
             body.put("data", dataRow);
             requestBody.put("json", body);
-            LOG.debug("当前发送的数据为:{}", HttpUtil.gson.toJson(requestBody));
+            LOG.debug("当前发送的数据为:{}", gson.toJson(requestBody));
             sendRequest(httpClient, requestBody, method, header, url);
         } catch (Exception e) {
             requestErrorMessage(e, index, row);
@@ -111,7 +116,7 @@ protected void writeMultipleRecordsInternal() throws Exception {
             }
             body.put("data", dataRow);
             requestBody.put("json", body);
-            LOG.debug("当前发送的数据为:{}", HttpUtil.gson.toJson(requestBody));
+            LOG.debug("当前发送的数据为:{}", gson.toJson(requestBody));
             sendRequest(httpClient, requestBody, method, header, url);
         } catch (Exception e) {
             LOG.error(ExceptionUtil.getErrorMessage(e));
@@ -134,7 +139,7 @@ private Object getDataFromRow(Row row, List<String> column) throws IOException {
             for (; index < row.getArity(); index++) {
                 columnData.put(column.get(index), row.getField(index));
             }
-            return HttpUtil.gson.toJson(columnData);
+            return gson.toJson(columnData);
         } else {
             return row.getField(index);
         }
@@ -146,7 +151,7 @@ private void sendRequest(CloseableHttpClient httpClient,
                              String method,
                              Map<String, String> header,
                              String url) throws IOException {
-        LOG.debug("当前发送的数据为:{}", HttpUtil.gson.toJson(requestBody));
+        LOG.debug("当前发送的数据为:{}", gson.toJson(requestBody));
         HttpRequestBase request = HttpUtil.getRequest(method, requestBody, header, url);
         //设置请求和传输超时时间
         RequestConfig requestConfig = RequestConfig.custom()