[INLONG-10520][Manager] Fix the problem of protocol type reported by …

…dataproxy not correctly accepted (apache#10521) * [INLONG-10520][Manager] Fix the problem of protocol type reported by dataproxy not correctly accepted * [INLONG-10520][Manager] Fix error * [INLONG-10520][Manager] Fix error
justinwwhuang · Jun 27, 2024 · 396ae72 · 396ae72
1 parent 147c49a
commit 396ae72
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 1 deletion.
diff --git a/docker/docker-compose/docker-compose.yml b/docker/docker-compose/docker-compose.yml
@@ -60,7 +60,7 @@ services:
       - ZK_URL=tubemq-server:2181
       - FLINK_HOST=jobmanager
       - FLINK_PORT=8081
-      - AUDIT_QUERY_URL=audit:10080
+      - AUDIT_QUERY_URL=http://audit:10080
 
   dashboard:
     image: inlong/dashboard:${VERSION_TAG}

diff --git a/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/HeartbeatMsg.java b/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/HeartbeatMsg.java
@@ -19,12 +19,14 @@
 
 import org.apache.inlong.common.enums.NodeSrvStatus;
 
+import com.google.common.base.Joiner;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
 import java.util.List;
+import java.util.stream.Collectors;
 
 /**
  * Heartbeat template for all components.
@@ -113,6 +115,14 @@ public class HeartbeatMsg {
     private Integer load = 0xffff;
 
     public ComponentHeartbeat componentHeartbeat() {
+        if (addressInfos != null) {
+            port = Joiner.on(",").join(addressInfos.stream().map(AddressInfo::getPort).collect(Collectors.toList()));
+            ip = Joiner.on(",").join(addressInfos.stream().map(AddressInfo::getIp).collect(Collectors.toList()));
+            reportSourceType = Joiner.on(",")
+                    .join(addressInfos.stream().map(AddressInfo::getReportSourceType).collect(Collectors.toList()));
+            protocolType = Joiner.on(",")
+                    .join(addressInfos.stream().map(AddressInfo::getProtocolType).collect(Collectors.toList()));
+        }
         return new ComponentHeartbeat(nodeSrvStatus, clusterTag, extTag, clusterName,
                 componentType, ip, port, inCharges, protocolType, load);
     }

diff --git a/...r-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java b/...r-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java
@@ -162,6 +162,8 @@ public void reportHeartbeat(HeartbeatMsg heartbeat) {
                     .join(heartbeat.getAddressInfos().stream().map(AddressInfo::getIp).collect(Collectors.toList())));
             heartbeat.setReportSourceType(Joiner.on(InlongConstants.COMMA).join(heartbeat.getAddressInfos().stream()
                     .map(AddressInfo::getReportSourceType).collect(Collectors.toList())));
+            heartbeat.setProtocolType(Joiner.on(InlongConstants.COMMA).join(heartbeat.getAddressInfos().stream()
+                    .map(AddressInfo::getProtocolType).collect(Collectors.toList())));
         }
 
         // protocolType may be null, and the protocolTypes' length may be less than ports' length