修改代码规范

Author: dengdejiushini

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch6/Elasticsearch6Sink.java

@@ -32,14 +32,14 @@
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.elasticsearch6.table.ElasticsearchTableInfo;
 import com.dtstack.flink.sql.table.TargetTableInfo;
+import com.google.common.collect.Maps;
 import org.apache.commons.lang.StringUtils;
 import org.apache.http.HttpHost;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -105,7 +105,7 @@ public TypeInformation<?>[] getFieldTypes() {
     private RichSinkFunction createEsSinkFunction() {
 
 
-        Map<String, String> userConfig = new HashMap<>();
+        Map<String, String> userConfig = Maps.newHashMap();
         userConfig.put("cluster.name", clusterName);
         // This instructs the sink to emit after every element, otherwise they would be buffered
         userConfig.put(org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "" + bulkFlushMaxActions);

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch6/Es6Util.java

@@ -21,6 +21,8 @@
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
 
+import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
+
 import com.dtstack.flink.sql.util.DtStringUtil;
 
 import java.util.HashMap;
@@ -35,7 +37,7 @@ public class Es6Util {
 
     public static Map<String, Object> rowToJsonMap(Row row, List<String> fields, List<String> types) {
         Preconditions.checkArgument(row.getArity() == fields.size());
-        Map<String,Object> jsonMap = new HashMap<>();
+        Map<String,Object> jsonMap = Maps.newHashMap();
         int i = 0;
         for(; i < fields.size(); ++i) {
             String field = fields.get(i);
@@ -44,7 +46,8 @@ public static Map<String, Object> rowToJsonMap(Row row, List<String> fields, Lis
             for(int j = 0; j < parts.length - 1; ++j) {
                 String key = parts[j];
                 if(currMap.get(key) == null) {
-                    currMap.put(key, new HashMap<String,Object>());
+                    HashMap<String, Object> hashMap = Maps.newHashMap();
+                    currMap.put(key, hashMap);
                 }
                 currMap = (Map<String, Object>) currMap.get(key);
             }

…asticsearch6/ExtendES6ApiCallBridge.java …asticsearch6/ExtendEs6ApiCallBridge.javaelasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch6/ExtendES6ApiCallBridge.java renamed to elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch6/ExtendEs6ApiCallBridge.java elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch6/ExtendES6ApiCallBridge.java renamed to elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch6/ExtendEs6ApiCallBridge.java

@@ -31,6 +31,7 @@
 import org.elasticsearch.action.bulk.BackoffPolicy;
 import org.elasticsearch.action.bulk.BulkItemResponse;
 import org.elasticsearch.action.bulk.BulkProcessor;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.client.RestClientBuilder;
 import org.elasticsearch.client.RestHighLevelClient;
@@ -48,17 +49,17 @@
  * @author yinxi
  * @date 2020/1/9 - 15:09
  */
-public class ExtendES6ApiCallBridge implements ElasticsearchApiCallBridge<RestHighLevelClient> {
+public class ExtendEs6ApiCallBridge implements ElasticsearchApiCallBridge<RestHighLevelClient> {
 
-    private static final Logger LOG = LoggerFactory.getLogger(ExtendES6ApiCallBridge.class);
+    private static final Logger LOG = LoggerFactory.getLogger(ExtendEs6ApiCallBridge.class);
 
     private final List<HttpHost> HttpAddresses;
 
     protected ElasticsearchTableInfo es6TableInfo;
 
-    public ExtendES6ApiCallBridge(List<HttpHost> HttpAddresses, ElasticsearchTableInfo es6TableInfo) {
-        Preconditions.checkArgument(HttpAddresses != null && !HttpAddresses.isEmpty());
-        this.HttpAddresses = HttpAddresses;
+    public ExtendEs6ApiCallBridge(List<HttpHost> httpAddresses, ElasticsearchTableInfo es6TableInfo) {
+        Preconditions.checkArgument(httpAddresses != null && !httpAddresses.isEmpty());
+        this.HttpAddresses = httpAddresses;
         this.es6TableInfo = es6TableInfo;
     }
 
@@ -81,7 +82,7 @@ public RestHighLevelClient createClient(Map<String, String> clientConfig) {
         }
 
         try{
-            if (!rhlClient.ping()) {
+            if (!rhlClient.ping(RequestOptions.DEFAULT)) {
                 throw new RuntimeException("There are no reachable Elasticsearch nodes!");
             }
         } catch (IOException e){

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch6/MetricElasticsearch6Sink.java

@@ -49,7 +49,7 @@ public class MetricElasticsearch6Sink<T> extends ElasticsearchSinkBase<T, RestHi
     public MetricElasticsearch6Sink(Map userConfig, List transportAddresses,
                                     ElasticsearchSinkFunction elasticsearchSinkFunction,
                                     ElasticsearchTableInfo es6TableInfo) {
-        super(new ExtendES6ApiCallBridge(transportAddresses, es6TableInfo), userConfig, elasticsearchSinkFunction, new NoOpFailureHandler());
+        super(new ExtendEs6ApiCallBridge(transportAddresses, es6TableInfo), userConfig, elasticsearchSinkFunction, new NoOpFailureHandler());
         this.customerSinkFunc = (CustomerSinkFunc) elasticsearchSinkFunction;
         this.userConfig = userConfig;
     }

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch6/table/ElasticsearchSinkParser.java

@@ -21,6 +21,7 @@
 import com.dtstack.flink.sql.table.AbsTableParser;
 import com.dtstack.flink.sql.table.TableInfo;
 import com.dtstack.flink.sql.util.MathUtil;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.Map;
 
@@ -46,6 +47,8 @@ public class ElasticsearchSinkParser extends AbsTableParser {
 
     private static final String KEY_ES6_PASSWORD = "password";
 
+    private static final String KEY_TRUE = "true";
+
     @Override
     protected boolean fieldNameNeedsUpperCase() {
         return false;
@@ -63,7 +66,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         elasticsearchTableInfo.setEsType((String) props.get(KEY_ES6_TYPE.toLowerCase()));
 
         String authMeshStr = (String) props.get(KEY_ES6_AUTHMESH.toLowerCase());
-        if (authMeshStr != null && "true".equalsIgnoreCase(authMeshStr)) {
+        if (authMeshStr != null && StringUtils.equalsIgnoreCase(KEY_TRUE, authMeshStr)) {
             elasticsearchTableInfo.setAuthMesh(MathUtil.getBoolean(authMeshStr));
             elasticsearchTableInfo.setUserName(MathUtil.getString(props.get(KEY_ES6_USERNAME.toLowerCase())));
             elasticsearchTableInfo.setPassword(MathUtil.getString(props.get(KEY_ES6_PASSWORD.toLowerCase())));