Merge branch 'bugfix/some-err' into 'main'

[GAL-444] fix: output unmatched sip record

See merge request galaxy/tsg_olap/sip-rtp-correlation!19

pom.xml

@@ -7,7 +7,7 @@
 
     <groupId>com.zdjizhi</groupId>
     <artifactId>sip-rtp-correlation</artifactId>
-    <version>1.1</version>
+    <version>1.2-rc2</version>
 
     <name>Flink : SIP-RTP : Correlation</name>
 

src/main/java/com/zdjizhi/flink/voip/CorrelateApp.java

@@ -59,7 +59,7 @@ public class CorrelateApp {
                                 .<ObjectNode>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                 .withTimestampAssigner((SerializableTimestampAssigner<ObjectNode>)
                                         (element, recordTimestamp) ->
-                                                element.get("common_start_timestamp_ms").asLong()));
+                                                element.get("start_timestamp_ms").asLong()));
 
         final ErrorHandler errorHandler = new ErrorHandler(config);
 

src/main/java/com/zdjizhi/flink/voip/error/ErrorHandler.java

@@ -9,6 +9,7 @@ import com.zdjizhi.flink.voip.records.SIPRecord;
 import com.zdjizhi.flink.voip.records.SchemaType;
 import com.zdjizhi.flink.voip.records.StreamDir;
 import com.zdjizhi.utils.IPUtil;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
@@ -21,8 +22,6 @@ import org.apache.flink.util.OutputTag;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.function.Function;
-
 /**
  * The ErrorHandler class is responsible for handling and filtering error records from the input data stream.
  * It checks for records that have invalid or meaningless addresses and ports, and outputs them to a separate stream if enabled.
@@ -109,13 +108,11 @@ class MeaninglessAddressProcessFunction extends ProcessFunction<ObjectNode, Obje
                                Collector<ObjectNode> out) throws Exception {
         final Record record = new Record(obj);
         // Check for invalid or meaningless addresses and ports
-        boolean cond1 = isIPAddress(record.getClientIp()) &&
-                isIPAddress(record.getServerIp()) &&
+        boolean cond1 = StringUtils.isNotBlank(record.getClientIp()) &&
+                StringUtils.isNotBlank(record.getServerIp()) &&
                 record.getClientPort() > 0 &&
                 record.getServerPort() > 0;
 
-        boolean cond8 = null != executeSafely(Record::getStreamDir, record);
-
         final SIPRecord sipRecord = new SIPRecord(obj);
         boolean cond2 = isIPAddress(sipRecord.getOriginatorSdpConnectIp())
                 || isIPAddress(sipRecord.getResponderSdpConnectIp());
@@ -124,16 +121,12 @@ class MeaninglessAddressProcessFunction extends ProcessFunction<ObjectNode, Obje
         boolean cond4 = !isIPAddress(sipRecord.getOriginatorSdpConnectIp())
                 || (includeIntranetIp || !isInternalIp(sipRecord.getOriginatorSdpConnectIp()));
         boolean cond5 = SchemaType.SIP.equals(sipRecord.getSchemaType());
-        boolean cond6 = StreamDir.DOUBLE == executeSafely(Record::getStreamDir, sipRecord) &&
+        boolean cond6 = StreamDir.DOUBLE == sipRecord.getStreamDir() &&
                 (includeIntranetIp || !isInternalIp(sipRecord.getResponderSdpConnectIp())) &&
                 (includeIntranetIp || !isInternalIp(sipRecord.getOriginatorSdpConnectIp()));
 
-        boolean cond7 = isIPAddress(sipRecord.getOriginatorSdpConnectIp()) &&
-                isIPAddress(sipRecord.getResponderSdpConnectIp()) &&
-                sipRecord.getResponderSdpMediaPort() > 0 && sipRecord.getOriginatorSdpMediaPort() > 0;
-
         // Both client and server addresses in the data are valid.
-        if (cond1 && cond8 && (!cond5 || cond7) && (
+        if (cond1 && (
                 // The address in the SIP one-way stream is valid and not an internal network address.
                 cond2 && cond3 && cond4 && cond5
                         // The coordinating addresses in the SIP double directional stream are valid
@@ -153,14 +146,6 @@ class MeaninglessAddressProcessFunction extends ProcessFunction<ObjectNode, Obje
     // ======================================================================================
     // ----------------------------------- private helper -----------------------------------
 
-    public static <T, R> R executeSafely(Function<T, R> function, T v) {
-        try {
-            return function.apply(v);
-        } catch (Exception e) {
-            return null;
-        }
-    }
-
     private static boolean isIPAddress(final String ipaddr) {
         if (null == ipaddr) {
             return false;

src/main/java/com/zdjizhi/flink/voip/functions/VoIPFusionFunction.java

@@ -83,7 +83,6 @@ public class VoIPFusionFunction extends KeyedCoProcessFunction<Tuple2<Integer, A
             rtpRecord.merge(sipObj)
                     .setString(Record.F_COMMON_SCHEMA_TYPE, SchemaType.VOIP.getValue());
             out.collect(rtpObj);
-            iterator.remove();
 
             switch (entry.getKey()) {
                 case S2C:

src/main/java/com/zdjizhi/flink/voip/records/Record.java

@@ -20,31 +20,35 @@ public class Record {
     /**
      * 字段名：数据记录中的所属 vsys
      */
-    public static final String F_COMMON_VSYS_ID = "common_vsys_id";
+    public static final String F_COMMON_VSYS_ID = "vsys_id";
     /**
      * 字段名：数据记录中的字段类型
      */
-    public static final String F_COMMON_SCHEMA_TYPE = "common_schema_type";
+    public static final String F_COMMON_SCHEMA_TYPE = "decoded_as";
     /**
      * 字段名：数据记录中的流类型
      */
     public static final String F_COMMON_STREAM_DIR = "common_stream_dir";
+    /**
+     * 字段名：数据记录中的流类型的 Flags
+     */
+    public static final String F_FLAGS = "flags";
     /**
      * 字段名：数据记录中的服务端地址
      */
-    public static final String F_COMMON_SERVER_IP = "common_server_ip";
+    public static final String F_COMMON_SERVER_IP = "server_ip";
     /**
      * 字段名：数据记录中的服务端端口
      */
-    public static final String F_COMMON_SERVER_PORT = "common_server_port";
+    public static final String F_COMMON_SERVER_PORT = "server_port";
     /**
      * 字段名：数据记录中的客户端地址
      */
-    public static final String F_COMMON_CLIENT_IP = "common_client_ip";
+    public static final String F_COMMON_CLIENT_IP = "client_ip";
     /**
      * 字段名：数据记录中的客户端端口
      */
-    public static final String F_COMMON_CLIENT_PORT = "common_client_port";
+    public static final String F_COMMON_CLIENT_PORT = "client_port";
 
     /**
      * ObjectNode data.
@@ -76,7 +80,7 @@ public class Record {
      * @return The stream direction.
      */
     public final StreamDir getStreamDir() {
-        return StreamDir.of(Record.getInt(obj, F_COMMON_STREAM_DIR));
+        return StreamDir.ofFlags(Record.getLong(obj, F_FLAGS));
     }
 
     /**
@@ -170,6 +174,30 @@ public class Record {
         return getInt(obj, field, 0);
     }
 
+    /**
+     * Gets a long value from the specified field in the ObjectNode.
+     *
+     * @param obj          The ObjectNode to get the value from.
+     * @param field        The name of the field.
+     * @param defaultValue The default value to return if the field is not found or is not a long.
+     * @return The long value from the field or the default value if the field is not found or is not a long.
+     */
+    public static long getLong(final ObjectNode obj, final String field, final long defaultValue) {
+        final JsonNode node = obj.get(field);
+        return node != null && node.isNumber() ? node.asLong() : defaultValue;
+    }
+
+    /**
+     * Gets a long value from the specified field in the ObjectNode.
+     *
+     * @param obj   The ObjectNode to get the value from.
+     * @param field The name of the field.
+     * @return The long value from the field or 0 if the field is not found or is not a long.
+     */
+    private static long getLong(final ObjectNode obj, final String field) {
+        return getLong(obj, field, 0L);
+    }
+
     /**
      * Get a string value from the specified field in the ObjectNode.
      *

src/main/java/com/zdjizhi/flink/voip/records/StreamDir.java

@@ -48,4 +48,22 @@ public enum StreamDir {
         }
         throw new IllegalArgumentException("Unknown StreamDir value '" + value + "'.");
     }
+
+    /**
+     * Get the StreamDir enum based on the provided flags value.
+     *
+     * @param flags The flags.
+     * @return The corresponding StreamDir enum.
+     * @throws IllegalArgumentException if the provided value does not match any known StreamDir.
+     */
+    public static StreamDir ofFlags(long flags) {
+        int v = 0;
+        if ((flags & 8192) == 8192) {
+            v += 1;
+        }
+        if ((flags & 16384) == 16384) {
+            v += 2;
+        }
+        return of(v);
+    }
 }