chore: update versions

[GAL-444] fix: fix error caused by invalid flags

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

README.md

@@ -22,18 +22,19 @@ 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                   |
-| sink.kafka.topic            | STRING              | Y |  | 将合成的 VoIP 及 未关联成功的 RTP 数据写出的 Kafka Topic 名 |
-| 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 关联的窗口大小（单位：分钟）                   |
+| 配置项                              | 类型                | 必需 | 默认值 | 描述                                        |
+|----------------------------------| ------------------- | ---------- | ---------------------------------------------------------- |-------------------------------------------|
+| 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      | 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 关联的窗口大小（单位：分钟）                  |
+| job.name                         | STRING | N | correlation_sip_rtp_session | Job 名                               |
 
 
 

pom.xml

@@ -7,7 +7,7 @@
 
     <groupId>com.zdjizhi</groupId>
     <artifactId>sip-rtp-correlation</artifactId>
-    <version>1.1</version>
+    <version>1.2.1</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);
 
@@ -103,6 +103,6 @@ public class CorrelateApp {
                 .union(sipDoubleDirOperator.getSideOutput(SIPPairingFunction.SIP_OUTPUT_TAG))
                 .addSink(producer);
 
-        env.execute("SIP-RTP-CORRELATION");
+        env.execute(config.get(JOB_NAME));
     }
 }

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

@@ -95,4 +95,13 @@ public class FusionConfigs {
                     .intType()
                     .defaultValue(6)
                     .withDescription("The interval at which RTP state data should be cleared.");
+
+    /**
+     * Configuration option for specifying the name of a job.
+     */
+    public static final ConfigOption<String> JOB_NAME =
+            ConfigOptions.key("job.name")
+                    .stringType()
+                    .defaultValue("correlation_sip_rtp_session")
+                    .withDescription("The name of current job.");
 }

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);
+    }
 }