拼接Protocol Statck ID时去除重复的基础协议。（TSG-18697）

src/main/java/com/zdjizhi/utils/functions/process/ParsingData.java

@@ -28,14 +28,7 @@ public class ParsingData extends ProcessFunction<String, Tuple3<Tags, Fields, Lo
                     Tags tags = JSONObject.parseObject(originalLog.getString("tags"), Tags.class);
                     Long timestamp_ms = originalLog.getLong("timestamp_ms");
 
-                    String appFullPath = tags.getApp_name();
+                    joinProtocol(tags);
-                    if (StringUtil.isNotBlank(appFullPath)) {
-                        String appName = appFullPath.substring(appFullPath.lastIndexOf(".") + 1);
-                        String protocolLabel = tags.getProtocol_stack_id();
-
-                        tags.setApp_name(appName);
-                        tags.setProtocol_stack_id(protocolLabel.concat(".").concat(appFullPath));
-                    }
 
                     out.collect(new Tuple3<>(tags, fields, timestamp_ms));
                 }
@@ -44,4 +37,32 @@ public class ParsingData extends ProcessFunction<String, Tuple3<Tags, Fields, Lo
             logger.error("Parsing application_protocol_stat data is abnormal! The exception message is: {}", e.getMessage());
         }
     }
+
+    /**
+     * 避免计算重复的协议，去除Decoded Path（最后一个元素） 与 Application（第一个元素）重复的基础协议。
+     *
+     * @param tags
+     */
+    private static void joinProtocol(Tags tags) {
+        String appFullPath = tags.getApp_name();
+
+        if (StringUtil.isNotBlank(appFullPath)) {
+            String appName = appFullPath.substring(appFullPath.lastIndexOf(".") + 1);
+            tags.setApp_name(appName);
+
+            String protocolLabel = tags.getProtocol_stack_id();
+
+            String endProtocol = protocolLabel.substring(protocolLabel.lastIndexOf(".") + 1);
+
+            String[] appSplits = appFullPath.split("\\.");
+
+            String firstAppProtocol = appSplits[0];
+
+            if (endProtocol.equals(firstAppProtocol)) {
+                tags.setProtocol_stack_id(protocolLabel.substring(0, protocolLabel.lastIndexOf(".")).concat(".").concat(appFullPath));
+            } else {
+                tags.setProtocol_stack_id(protocolLabel.concat(".").concat(appFullPath));
+            }
+        }
+    }
 }

src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java

@@ -15,7 +15,7 @@ import java.util.Properties;
 public class KafkaProducer {
 
     public static FlinkKafkaProducer<String> getKafkaProducer(Properties properties, String topic, boolean logFailuresOnly) {
-        setDefaultConfig(properties, "ack", 1);
+        setDefaultConfig(properties, "ack", "1");
         setDefaultConfig(properties, "retries", 0);
         setDefaultConfig(properties, "linger.ms", 10);
         setDefaultConfig(properties, "request.timeout.ms", 30000);

src/test/java/com/zdjizhi/ConventionalTest.java

@@ -72,4 +72,26 @@ public class ConventionalTest {
         System.out.println(protocol.concat(".").concat(app));
 
     }
+
+    @Test
+    public void removeDuplicateProtocolTest() {
+        String str = "[.]";
+        String protocol = "ETHERNET.IPv4.TCP.http";
+        String[] protocolSplits = protocol.split(str);
+        String endProtocol = protocolSplits[protocolSplits.length -1];
+        System.out.println(endProtocol);
+
+        String app = "http.test";
+        String[] appSplits = app.split(str);
+        String firstAppProtocol = appSplits[0];
+        System.out.println(firstAppProtocol);
+
+
+        System.out.println(app.substring(app.lastIndexOf(".") + 1));
+        System.out.println(app.lastIndexOf(".") + 1);
+        System.out.println(protocol.substring(0,protocol.lastIndexOf(".")));
+
+    }
+
+
 }