From 0acc85e67b3a21c73aa760f3690de28945ea4c8b Mon Sep 17 00:00:00 2001
From: "LAPTOP-CUUVN8AS\\wk" <www.572159764@qq.com>
Date: Tue, 5 Jul 2022 18:36:48 +0800
Subject: [PATCH] =?UTF-8?q?radius=E4=B8=8B=E7=BA=BF=E7=9A=84=E5=8A=9F?=
 =?UTF-8?q?=E8=83=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pom.xml                                             |  6 +++++-
 properties/service_flow_config.properties           |  4 ++--
 src/main/java/com/zdjizhi/pojo/RadiusMassage.java   | 13 ++++++++++---
 .../java/com/zdjizhi/topology/RadiusRelation.java   |  2 +-
 .../com/zdjizhi/utils/functions/ParseFunction.java  |  7 +++++--
 .../utils/hbasepackage/HbaseSinkAccount.java        |  2 +-
 .../utils/hbasepackage/HbaseSinkFramedip.java       |  3 ++-
 7 files changed, 26 insertions(+), 11 deletions(-)

diff --git a/pom.xml b/pom.xml
index b89be0a..9c4b862 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,7 +6,11 @@
 
     <groupId>com.zdjizhi</groupId>
     <artifactId>radius-relation</artifactId>
+<<<<<<< HEAD
+    <version>22-04-01</version>
+=======
     <version>22-03-09</version>
+>>>>>>> c8041564036f715f1a9eacc8bf6deb513542b2c6
 
     <name>radius-relation</name>
     <url>http://www.example.com</url>
@@ -251,7 +255,7 @@
                             <goal>shade</goal>
                         </goals>
                         <configuration>
-                            <finalName>radius-relation-21-12-06</finalName>
+                            <finalName>radius-relation-22-04-01</finalName>
                             <transformers combine.children="append">
                                 <!-- The service transformer is needed to merge META-INF/services files -->
                                 <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
diff --git a/properties/service_flow_config.properties b/properties/service_flow_config.properties
index 15aca81..ce042b7 100644
--- a/properties/service_flow_config.properties
+++ b/properties/service_flow_config.properties
@@ -2,11 +2,11 @@
 
 #管理kafka地址
 #input.kafka.servers=192.168.44.11:9094,192.168.44.14:9094,192.168.44.15:9094
-input.kafka.servers=192.168.44.85:9094
+input.kafka.servers=192.168.44.12:9094
 
 #hbase zookeeper地址 用于连接HBase
 #hbase.zookeeper.servers=192.168.44.12
-hbase.zookeeper.servers=192.168.44.85:2181
+hbase.zookeeper.servers=192.168.44.12:2181
 
 #--------------------------------Kafka消费组信息------------------------------#
 
diff --git a/src/main/java/com/zdjizhi/pojo/RadiusMassage.java b/src/main/java/com/zdjizhi/pojo/RadiusMassage.java
index 102211d..97377dc 100644
--- a/src/main/java/com/zdjizhi/pojo/RadiusMassage.java
+++ b/src/main/java/com/zdjizhi/pojo/RadiusMassage.java
@@ -4,9 +4,10 @@ public class RadiusMassage {
 
     private String radius_framed_ip;
     private String radius_account;
-    private Long radius_event_timestamp;
+    private long radius_event_timestamp;
     private int radius_acct_status_type;
     private int radius_packet_type;
+    private long common_end_time;
 
     public int getRadius_acct_status_type() {
         return radius_acct_status_type;
@@ -40,13 +41,19 @@ public class RadiusMassage {
         this.radius_account = radius_account;
     }
 
-    public Long getRadius_event_timestamp() {
+    public long getRadius_event_timestamp() {
         return radius_event_timestamp;
     }
 
-    public void setRadius_event_timestamp(Long radius_event_timestamp) {
+    public void setRadius_event_timestamp(long radius_event_timestamp) {
         this.radius_event_timestamp = radius_event_timestamp;
     }
 
+    public long getCommon_end_time() {
+        return common_end_time;
+    }
 
+    public void setCommon_end_time(long common_end_time) {
+        this.common_end_time = common_end_time;
+    }
 }
diff --git a/src/main/java/com/zdjizhi/topology/RadiusRelation.java b/src/main/java/com/zdjizhi/topology/RadiusRelation.java
index 352e920..10e9127 100644
--- a/src/main/java/com/zdjizhi/topology/RadiusRelation.java
+++ b/src/main/java/com/zdjizhi/topology/RadiusRelation.java
@@ -41,7 +41,7 @@ public class RadiusRelation {
 
         accountWithFrameip.addSink(new HbaseSinkAccount(RadiusRelationshipConfig.HBASE_ZOOKEEPER_SERVERS));
         try {
-            environment.execute("RADIUS-RELATIONSHIP-HBASE-V2-t");
+            environment.execute("RADIUS-RELATIONSHIP-HBASE-V2");
         } catch (Exception e) {
             logger.error("This Flink task start ERROR! Exception information is :" + e);
         }
diff --git a/src/main/java/com/zdjizhi/utils/functions/ParseFunction.java b/src/main/java/com/zdjizhi/utils/functions/ParseFunction.java
index cf9515f..21506ce 100644
--- a/src/main/java/com/zdjizhi/utils/functions/ParseFunction.java
+++ b/src/main/java/com/zdjizhi/utils/functions/ParseFunction.java
@@ -33,12 +33,15 @@ public class ParseFunction implements MapFunction<String, Tuple6<String,String,S
                 if (StringUtil.isNotBlank(message)) {
                     radiusMassage = JSON.parseObject(message, RadiusMassage.class);
 
-                    if(radiusMassage.getRadius_framed_ip()!=null && radiusMassage.getRadius_account()!=null && radiusMassage.getRadius_event_timestamp()!=null){
+                        if(radiusMassage.getRadius_framed_ip()!=null && radiusMassage.getRadius_account()!=null ){
 
                         if (RadiusRelationshipConfig.ACCOUNTING_REQUEST == radiusMassage.getRadius_packet_type()){
                             String framedIp=radiusMassage.getRadius_framed_ip();
                             String account=radiusMassage.getRadius_account();
-                            Long event_time = radiusMassage.getRadius_event_timestamp();
+                            long event_time = radiusMassage.getRadius_event_timestamp();
+                            if(event_time==0){
+                                event_time=radiusMassage.getCommon_end_time();
+                            }
                             int status =radiusMassage.getRadius_acct_status_type();
                             int onff_status = 1;
                             if (status == 2) {
diff --git a/src/main/java/com/zdjizhi/utils/hbasepackage/HbaseSinkAccount.java b/src/main/java/com/zdjizhi/utils/hbasepackage/HbaseSinkAccount.java
index 44fd694..515f16e 100644
--- a/src/main/java/com/zdjizhi/utils/hbasepackage/HbaseSinkAccount.java
+++ b/src/main/java/com/zdjizhi/utils/hbasepackage/HbaseSinkAccount.java
@@ -93,7 +93,7 @@ public class HbaseSinkAccount extends RichSinkFunction<Tuple6<String,String,Stri
         if (AccountWithIpMap.containsKey(key)) {
             Radius radius = AccountWithIpMap.get(key);
 
-            if (radius.getLast_update_time() < event_time) {
+            if (radius.getLast_update_time() <= event_time) {
                 if(acct_status_type==radius.getAcct_status_type()) {
 
                     if (!radius.getFramed_ip().equals(framedIp)) {
diff --git a/src/main/java/com/zdjizhi/utils/hbasepackage/HbaseSinkFramedip.java b/src/main/java/com/zdjizhi/utils/hbasepackage/HbaseSinkFramedip.java
index bcb608a..598b82d 100644
--- a/src/main/java/com/zdjizhi/utils/hbasepackage/HbaseSinkFramedip.java
+++ b/src/main/java/com/zdjizhi/utils/hbasepackage/HbaseSinkFramedip.java
@@ -94,7 +94,7 @@ public class HbaseSinkFramedip extends RichSinkFunction<Tuple6<String,String,Str
             if (IpWithAccountMap.containsKey(key)) {
                 Radius radius = IpWithAccountMap.get(key);
 
-                if (radius.getLast_update_time() < event_time) {
+                if (radius.getLast_update_time() <= event_time) {
 
                     if(acct_status_type==radius.getAcct_status_type()) {
 
@@ -164,6 +164,7 @@ public class HbaseSinkFramedip extends RichSinkFunction<Tuple6<String,String,Str
                     put.addColumn("radius".getBytes(), "first_found_time".getBytes(), Bytes.toBytes(event_time));
                     put.addColumn("radius".getBytes(), "acct_status_type".getBytes(), Bytes.toBytes(acct_status_type));
                     table.put(put);
+
                     IpWithAccountMap.put(key, radius);
                 } catch (Exception e) {
                     log.error(e.toString());