[fix-31977][metadatatidb][restapi] 解决在库名带.时切换失败的问题

Author: lvyanquan

flinkx-metadata-tidb/flinkx-metadata-tidb-reader/src/main/java/com/dtstack/flinkx/metadatatidb/constants/TidbMetadataCons.java

@@ -53,6 +53,7 @@ public class TidbMetadataCons extends MetaDataCons {
     public static final String RESULT_COMMENT = "Comment";
 
     /** sql语句 */
+    public static final String SQL_SWITCH_DATABASE = "USE `%s`";
     public static final String SQL_SHOW_TABLES = "SHOW FULL TABLES WHERE Table_type = 'BASE TABLE'";
     public static final String SQL_QUERY_TABLE_INFO = "SHOW TABLE STATUS LIKE '%s'";
     public static final String SQL_QUERY_COLUMN = "SHOW FULL COLUMNS FROM %s";

flinkx-restapi/flinkx-restapi-core/src/main/java/com/dtstack/flinkx/restapi/common/RestapiKeys.java

@@ -29,4 +29,5 @@ public class RestapiKeys {
     public static final String KEY_COLUMN = "column";
     public static final String KEY_URL = "url";
     public static final String KEY_BATCH_INTERVAL = "batchInterval";
+    public static final String KEY_BATCH = "batchId";
 }

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

@@ -38,6 +38,8 @@
 import java.util.Map;
 import java.util.UUID;
 
+import static com.dtstack.flinkx.restapi.common.RestapiKeys.KEY_BATCH;
+
 /**
  * @author : tiezhu
  * @date : 2020/3/12
@@ -73,9 +75,10 @@ protected void writeSingleRecordInternal(Row row) throws WriteRecordException {
         CloseableHttpClient httpClient = HttpUtil.getHttpClient();
         int index = 0;
         Map<String, Object> requestBody = Maps.newHashMap();
-        Object dataRow;
+        List<Object> dataRow = new ArrayList<>();
         try {
-            dataRow = getDataFromRow(row, column);
+            dataRow.add(getDataFromRow(row, column));
+            params.put(KEY_BATCH, UUID.randomUUID().toString().substring(0, 8));
             if (!params.isEmpty()) {
                 Iterator iterator = params.entrySet().iterator();
                 while (iterator.hasNext()) {
@@ -85,7 +88,7 @@ protected void writeSingleRecordInternal(Row row) throws WriteRecordException {
             }
             body.put("data", dataRow);
             requestBody.put("json", body);
-            LOG.debug("当前发送的数据为:{}", gson.toJson(requestBody));
+            LOG.info("send data:{}", gson.toJson(requestBody));
             sendRequest(httpClient, requestBody, method, header, url);
         } catch (Exception e) {
             requestErrorMessage(e, index, row);
@@ -107,6 +110,7 @@ protected void writeMultipleRecordsInternal() throws Exception {
             for (Row row : rows) {
                 dataRow.add(getDataFromRow(row, column));
             }
+            params.put(KEY_BATCH, UUID.randomUUID().toString().substring(0, 8));
             if (!params.isEmpty()) {
                 Iterator iterator = params.entrySet().iterator();
                 while (iterator.hasNext()) {
@@ -116,7 +120,7 @@ protected void writeMultipleRecordsInternal() throws Exception {
             }
             body.put("data", dataRow);
             requestBody.put("json", body);
-            LOG.debug("当前发送的数据为:{}", gson.toJson(requestBody));
+            LOG.info("this batch size = {}, send data:{}", rows.size(), gson.toJson(requestBody));
             sendRequest(httpClient, requestBody, method, header, url);
         } catch (Exception e) {
             LOG.error(ExceptionUtil.getErrorMessage(e));
@@ -151,7 +155,7 @@ private void sendRequest(CloseableHttpClient httpClient,
                              String method,
                              Map<String, String> header,
                              String url) throws IOException {
-        LOG.debug("当前发送的数据为:{}", gson.toJson(requestBody));
+        LOG.debug("send data:{}", gson.toJson(requestBody));
         HttpRequestBase request = HttpUtil.getRequest(method, requestBody, header, url);
         //设置请求和传输超时时间
         RequestConfig requestConfig = RequestConfig.custom()