Merge branch 'hotfix/output-sip' into 'release/1.0'

hotfix: add sip record output

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

CHANGELOG.md

@@ -0,0 +1,7 @@
+# Changelog
+
+### Hotfix
+ - [#5](https://git.mesalab.cn/galaxy/tsg_olap/sip-rtp-correlation/-/issues/5) 修复了由于 IPUtil 在判断 Ipv6 地址没有判空而引起的空指针异常
+
+### Other
+ - 输出 SIP Record

pom.xml

@@ -7,7 +7,7 @@
 
     <groupId>com.zdjizhi</groupId>
     <artifactId>sip-rtp-correlation</artifactId>
-    <version>1.0-rc1</version>
+    <version>1.0-rc4</version>
 
     <name>Flink : SIP-RTP : Correlation</name>
 
@@ -285,6 +285,20 @@
                 </executions>
             </plugin>
 
+            <plugin>
+                <groupId>io.github.zlika</groupId>
+                <artifactId>reproducible-build-maven-plugin</artifactId>
+                <version>0.2</version>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>strip-jar</goal>
+                        </goals>
+                        <phase>package</phase>
+                    </execution>
+                </executions>
+            </plugin>
+
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-shade-plugin</artifactId>

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

@@ -4,8 +4,8 @@ import com.zdjizhi.flink.voip.conf.FusionConfiguration;
 import com.zdjizhi.flink.voip.error.ErrorHandler;
 import com.zdjizhi.flink.voip.formats.JsonNodeSerializationSchema;
 import com.zdjizhi.flink.voip.functions.*;
-import com.zdjizhi.flink.voip.records.Record;
-import com.zdjizhi.flink.voip.records.SIPRecord;
+import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.java.functions.KeySelector;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.utils.ParameterTool;
@@ -13,12 +13,13 @@ import org.apache.flink.configuration.Configuration;
 import org.apache.flink.formats.json.JsonNodeDeserializationSchema;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
 import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
 
+import java.time.Duration;
+
 import static com.zdjizhi.flink.voip.conf.FusionConfigs.*;
 
 /**
@@ -52,7 +53,13 @@ public class CorrelateApp {
                 fusionConfiguration
                         .getProperties(SOURCE_KAFKA_PROPERTIES_PREFIX));
 
-        final DataStreamSource<ObjectNode> sourceStream = env.addSource(kafkaConsumer);
+        final DataStream<ObjectNode> sourceStream = env.addSource(kafkaConsumer)
+                .assignTimestampsAndWatermarks(
+                        WatermarkStrategy
+                                .<ObjectNode>forBoundedOutOfOrderness(Duration.ofSeconds(5))
+                                .withTimestampAssigner((SerializableTimestampAssigner<ObjectNode>)
+                                        (element, recordTimestamp) ->
+                                                element.get("common_start_timestamp_ms").asLong()));
 
         final ErrorHandler errorHandler = new ErrorHandler(config);
 
@@ -92,8 +99,8 @@ public class CorrelateApp {
                 new JsonNodeSerializationSchema(),
                 fusionConfiguration.getProperties(SINK_KAFKA_PROPERTIES_PREFIX));
 
-        voIpOperator.addSink(producer);
+        voIpOperator.union(sipDoubleDirOperator).addSink(producer);
 
-        env.execute("VoIP Fusion Job");
+        env.execute("SIP-RTP-CORRELATION");
     }
 }

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

@@ -104,16 +104,16 @@ class MeaninglessAddressProcessFunction extends ProcessFunction<ObjectNode, Obje
                 record.getServerPort() >= 0;
 
         final SIPRecord sipRecord = new SIPRecord(obj);
-        boolean cond2 = !IPUtil.isIPAddress(sipRecord.getOriginatorSdpConnectIp())
-                || IPUtil.internalIp(sipRecord.getOriginatorSdpConnectIp());
-        boolean cond3 = !IPUtil.isIPAddress(sipRecord.getResponderSdpConnectIp())
-                || IPUtil.internalIp(sipRecord.getResponderSdpConnectIp());
-        boolean cond4 = IPUtil.isIPAddress(sipRecord.getOriginatorSdpConnectIp())
-                || IPUtil.isIPAddress(sipRecord.getResponderSdpConnectIp());
+        boolean cond2 = !isIPAddress(sipRecord.getOriginatorSdpConnectIp())
+                || isInternalIp(sipRecord.getOriginatorSdpConnectIp());
+        boolean cond3 = !isIPAddress(sipRecord.getResponderSdpConnectIp())
+                || isInternalIp(sipRecord.getResponderSdpConnectIp());
+        boolean cond4 = isIPAddress(sipRecord.getOriginatorSdpConnectIp())
+                || isIPAddress(sipRecord.getResponderSdpConnectIp());
         boolean cond5 = SchemaType.SIP.equals(sipRecord.getSchemaType());
         boolean cond6 = StreamDir.DOUBLE == sipRecord.getStreamDir() &&
-                IPUtil.internalIp(sipRecord.getResponderSdpConnectIp()) &&
-                IPUtil.internalIp(sipRecord.getOriginatorSdpConnectIp());
+                isInternalIp(sipRecord.getResponderSdpConnectIp()) &&
+                isInternalIp(sipRecord.getOriginatorSdpConnectIp());
 
         // Both client and server addresses in the data are valid.
         if (cond1 && (
@@ -132,4 +132,21 @@ class MeaninglessAddressProcessFunction extends ProcessFunction<ObjectNode, Obje
             }
         }
     }
+
+    // ======================================================================================
+    // ----------------------------------- private helper -----------------------------------
+
+    private static boolean isIPAddress(final String ipaddr) {
+        if (null == ipaddr) {
+            return false;
+        }
+        return IPUtil.isIPAddress(ipaddr);
+    }
+
+    private static boolean isInternalIp(final String ipaddr) {
+        if (!isIPAddress(ipaddr)) {
+            return false;
+        }
+        return IPUtil.internalIp(ipaddr);
+    }
 }

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

@@ -26,7 +26,7 @@ public interface FunctionHelper extends RichFunction {
     }
 
     default void registerNextFireTimestamp(TimerService timerService, long interval) {
-        long current = timerService.currentProcessingTime();
-        timerService.registerProcessingTimeTimer(current + interval);
+        long current = timerService.currentWatermark();
+        timerService.registerEventTimeTimer(current + interval);
     }
 }