From 40b3751990808b8b12d1a7cef5fc18a02cc2d525 Mon Sep 17 00:00:00 2001
From: qidaijie <qidaijie@mesalab.cn>
Date: Thu, 18 Jan 2024 16:22:24 +0800
Subject: [PATCH] =?UTF-8?q?=E6=8B=BC=E6=8E=A5Protocol=20Statck=20ID?=
 =?UTF-8?q?=E6=97=B6=E5=8E=BB=E9=99=A4=E9=87=8D=E5=A4=8D=E7=9A=84=E5=9F=BA?=
 =?UTF-8?q?=E7=A1=80=E5=8D=8F=E8=AE=AE=E3=80=82=EF=BC=88TSG-18697=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../utils/functions/process/ParsingData.java  | 37 +++++++++++++++----
 .../zdjizhi/utils/kafka/KafkaProducer.java    |  2 +-
 .../java/com/zdjizhi/ConventionalTest.java    | 22 +++++++++++
 3 files changed, 52 insertions(+), 9 deletions(-)

diff --git a/src/main/java/com/zdjizhi/utils/functions/process/ParsingData.java b/src/main/java/com/zdjizhi/utils/functions/process/ParsingData.java
index 257915c..e88ae78 100644
--- a/src/main/java/com/zdjizhi/utils/functions/process/ParsingData.java
+++ b/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();
-                    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));
-                    }
+                    joinProtocol(tags);
 
                     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));
+            }
+        }
+    }
 }
diff --git a/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java b/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java
index c7cd3f2..c276dc8 100644
--- a/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java
+++ b/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);
diff --git a/src/test/java/com/zdjizhi/ConventionalTest.java b/src/test/java/com/zdjizhi/ConventionalTest.java
index 287b3fb..9e5f185 100644
--- a/src/test/java/com/zdjizhi/ConventionalTest.java
+++ b/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(".")));
+
+    }
+
+
 }