Merge branch 'release/1.1' into 'main'

chore: update version

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

CHANGELOG.md

@@ -3,5 +3,6 @@
 ### Hotfix
  - [#5](https://git.mesalab.cn/galaxy/tsg_olap/sip-rtp-correlation/-/issues/5) 修复了由于 IPUtil 在判断 Ipv6 地址没有判空而引起的空指针异常
 
-### Other
- - 输出 SIP Record
+### Feature
+ - 输出 SIP Record
+ - [GAL-419](https://jira.geedge.net/browse/GAL-419) 增加配置项 `include.intranet.ip`, 是否对 SIP 协商主叫 IP 或被叫 IP 为内网地址的数据进行关联，不关联则输出到异常 Topic 中。

README.md

@@ -22,17 +22,18 @@ flink run -c com.zdjizhi.flink.voip.CorrelateApp path/to/sip-rtp-correlation-<ve
 
 ## 配置项说明
 
-| 配置项                      | 类型                | 必需 | 默认值 | 描述                                                        |
-| --------------------------- | ------------------- | ---------- | ----------------------------------------------------------- | ----------------------------------------------------------- |
-| source.kafka.topic          | STRING              | Y |  | 将要读取的 Kafka Topic 名称，其包含 SIP 和 RTP 原始数据     |
-| source.kafka.props.*        | MAP<STRING, STRING> | Y |  | 将要读取的 Kafka 的 Properties                              |
+| 配置项                      | 类型                | 必需 | 默认值 | 描述                                         |
+| --------------------------- | ------------------- | ---------- | ----------------------------------------------------------- |--------------------------------------------|
+| source.kafka.topic          | STRING              | Y |  | 将要读取的 Kafka Topic 名称，其包含 SIP 和 RTP 原始数据    |
+| source.kafka.props.*        | MAP<STRING, STRING> | Y |  | 将要读取的 Kafka 的 Properties                   |
 | sink.kafka.topic            | STRING              | Y |  | 将合成的 VoIP 及 未关联成功的 RTP 数据写出的 Kafka Topic 名 |
-| sink.kafka.props.*          | MAP<STRING, STRING> | Y |  | 数据输出的 Kafka 的 Properties                              |
-| error.records.output.enable | STRING              | N | False | 是否开启异常数据的输出 【IP 或 Port 为空】                  |
-| error.sink.kafka.topic      | STRING              | N |      | 异常数据输出到的 Kafka Topic 名                             |
-| error.sink.kafka.props.*    | MAP<STRING, STRING> | N |      | 异常数据输出的 Kafka 的 Properties                          |
-| sip.state.clear.interval.minutes | INT | N | 1 | SIP 单向流关联的窗口大小（单位：分钟） |
-| rtp.state.clear.interval.minutes | INT | N | 6 | SIP 和 RTP 关联的窗口大小（单位：分钟） |
+| sink.kafka.props.*          | MAP<STRING, STRING> | Y |  | 数据输出的 Kafka 的 Properties                   |
+| error.records.output.enable | BOOLEAN   | N | False | 是否开启异常数据的输出 【IP 或 Port 为空】                 |
+| include.intranet.ip | BOOLEAN | N | True | 是否对 SIP 协商主叫 IP 或被叫 IP 为内网地址的数据进行关联        |
+| error.sink.kafka.topic      | STRING              | N |      | 异常数据输出到的 Kafka Topic 名                     |
+| error.sink.kafka.props.*    | MAP<STRING, STRING> | N |      | 异常数据输出的 Kafka 的 Properties                 |
+| sip.state.clear.interval.minutes | INT | N | 1 | SIP 单向流关联的窗口大小（单位：分钟）                      |
+| rtp.state.clear.interval.minutes | INT | N | 6 | SIP 和 RTP 关联的窗口大小（单位：分钟）                   |
 
 
 

pom.xml

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

src/main/java/com/zdjizhi/flink/voip/conf/FusionConfigs.java

@@ -56,6 +56,16 @@ public class FusionConfigs {
                     .withDescription("Enable or disable the output of error records. " +
                             "If set to true, the error records will be sent to the specified Kafka topic.");
 
+    /**
+     * Configuration option to determine whether to perform data correlate for intranet addresses.
+     */
+    public static final ConfigOption<Boolean> INCLUDE_INTRANET_IP =
+            ConfigOptions.key("include.intranet.ip")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription("Whether to perform data correlate for intranet addresses");
+
+
     /**
      * Configuration option for specifying the Kafka topic name where the error data will be sent.
      * This configuration option is used when the output of error records is enabled.

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

@@ -3,6 +3,7 @@ package com.zdjizhi.flink.voip.error;
 import com.zdjizhi.flink.voip.conf.FusionConfigs;
 import com.zdjizhi.flink.voip.conf.FusionConfiguration;
 import com.zdjizhi.flink.voip.formats.JsonNodeSerializationSchema;
+import com.zdjizhi.flink.voip.functions.FunctionHelper;
 import com.zdjizhi.flink.voip.records.Record;
 import com.zdjizhi.flink.voip.records.SIPRecord;
 import com.zdjizhi.flink.voip.records.SchemaType;
@@ -88,10 +89,19 @@ public class ErrorHandler {
  * The MeaninglessAddressProcessFunction is a custom ProcessFunction used to check for records
  * with invalid or meaningless addresses and ports. It separates them into the invalid output tag if necessary.
  */
-class MeaninglessAddressProcessFunction extends ProcessFunction<ObjectNode, ObjectNode> {
+class MeaninglessAddressProcessFunction extends ProcessFunction<ObjectNode, ObjectNode> implements FunctionHelper {
 
     private static final Logger LOG = LoggerFactory.getLogger(MeaninglessAddressProcessFunction.class);
 
+    private transient boolean includeIntranetIp;
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        final Configuration config = getGlobalConfiguration();
+        includeIntranetIp = config.get(FusionConfigs.INCLUDE_INTRANET_IP);
+    }
+
     @Override
     public void processElement(ObjectNode obj,
                                ProcessFunction<ObjectNode, ObjectNode>.Context ctx,
@@ -100,20 +110,20 @@ class MeaninglessAddressProcessFunction extends ProcessFunction<ObjectNode, Obje
         // Check for invalid or meaningless addresses and ports
         boolean cond1 = StringUtils.isNotBlank(record.getClientIp()) &&
                 StringUtils.isNotBlank(record.getServerIp()) &&
-                record.getClientPort() >= 0 &&
-                record.getServerPort() >= 0;
+                record.getClientPort() > 0 &&
+                record.getServerPort() > 0;
 
         final SIPRecord sipRecord = new SIPRecord(obj);
-        boolean cond2 = !isIPAddress(sipRecord.getOriginatorSdpConnectIp())
-                || isInternalIp(sipRecord.getOriginatorSdpConnectIp());
-        boolean cond3 = !isIPAddress(sipRecord.getResponderSdpConnectIp())
-                || isInternalIp(sipRecord.getResponderSdpConnectIp());
-        boolean cond4 = isIPAddress(sipRecord.getOriginatorSdpConnectIp())
+        boolean cond2 = isIPAddress(sipRecord.getOriginatorSdpConnectIp())
                 || isIPAddress(sipRecord.getResponderSdpConnectIp());
+        boolean cond3 = !isIPAddress(sipRecord.getResponderSdpConnectIp())
+                || (includeIntranetIp || !isInternalIp(sipRecord.getResponderSdpConnectIp()));
+        boolean cond4 = !isIPAddress(sipRecord.getOriginatorSdpConnectIp())
+                || (includeIntranetIp || !isInternalIp(sipRecord.getOriginatorSdpConnectIp()));
         boolean cond5 = SchemaType.SIP.equals(sipRecord.getSchemaType());
         boolean cond6 = StreamDir.DOUBLE == sipRecord.getStreamDir() &&
-                isInternalIp(sipRecord.getResponderSdpConnectIp()) &&
-                isInternalIp(sipRecord.getOriginatorSdpConnectIp());
+                (includeIntranetIp || !isInternalIp(sipRecord.getResponderSdpConnectIp())) &&
+                (includeIntranetIp || !isInternalIp(sipRecord.getOriginatorSdpConnectIp()));
 
         // Both client and server addresses in the data are valid.
         if (cond1 && (