This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 45e6cea21 [INLONG-8121][Manager] Supports cluster node status
management in the case of multiple manager nodes (#8124)
45e6cea21 is described below
commit 45e6cea2175b3a8d752402190916e35ace1d3a68
Author: fuweng11 <[email protected]>
AuthorDate: Tue May 30 12:28:32 2023 +0800
[INLONG-8121][Manager] Supports cluster node status management in the case
of multiple manager nodes (#8124)
---
.../manager/dao/mapper/ComponentHeartbeatEntityMapper.java | 10 ++++++++++
.../resources/mappers/ComponentHeartbeatEntityMapper.xml | 9 +++++++++
.../inlong/manager/service/heartbeat/HeartbeatManager.java | 13 +++++++++++++
.../src/main/resources/application-dev.properties | 2 +-
.../src/main/resources/application-prod.properties | 2 +-
.../src/main/resources/application-test.properties | 2 +-
6 files changed, 35 insertions(+), 3 deletions(-)
diff --git
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/ComponentHeartbeatEntityMapper.java
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/ComponentHeartbeatEntityMapper.java
index 2c35006bd..49607486e 100644
---
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/ComponentHeartbeatEntityMapper.java
+++
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/ComponentHeartbeatEntityMapper.java
@@ -38,6 +38,16 @@ public interface ComponentHeartbeatEntityMapper {
List<ComponentHeartbeatEntity> selectByCondition(@Param("request")
HeartbeatPageRequest request);
+ /**
+ * Get the heartbeat by heartbeat interval
+ *
+ * @param component component type
+ * @param instance component address
+ * @param beforeSeconds the modified time was beforeSeconds seconds ago
+ */
+ ComponentHeartbeatEntity selectTimeOutHeartBeat(@Param("component") String
component,
+ @Param("instance") String instance, @Param("beforeSeconds") Long
beforeSeconds);
+
int deleteByPrimaryKey(Integer id);
}
\ No newline at end of file
diff --git
a/inlong-manager/manager-dao/src/main/resources/mappers/ComponentHeartbeatEntityMapper.xml
b/inlong-manager/manager-dao/src/main/resources/mappers/ComponentHeartbeatEntityMapper.xml
index 6046e82dc..077ae3efd 100644
---
a/inlong-manager/manager-dao/src/main/resources/mappers/ComponentHeartbeatEntityMapper.xml
+++
b/inlong-manager/manager-dao/src/main/resources/mappers/ComponentHeartbeatEntityMapper.xml
@@ -77,6 +77,15 @@
order by modify_time desc
</select>
+ <select id="selectTimeOutHeartBeat"
resultType="org.apache.inlong.manager.dao.entity.ComponentHeartbeatEntity">
+ select
+ <include refid="Base_Column_List"/>
+ from component_heartbeat
+ where component = #{component, jdbcType=VARCHAR}
+ and instance = #{instance,jdbcType=VARCHAR}
+ and modify_time >= DATE_ADD(NOW(), INTERVAL -#{beforeSeconds,
jdbcType=INTEGER} SECOND)
+ </select>
+
<delete id="deleteByPrimaryKey" parameterType="java.lang.Integer">
delete
from component_heartbeat
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java
index c7d20cdcf..894a2df0a 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java
@@ -28,8 +28,10 @@ import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.NodeStatus;
import org.apache.inlong.manager.common.util.JsonUtils;
import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.dao.entity.ComponentHeartbeatEntity;
import org.apache.inlong.manager.dao.entity.InlongClusterEntity;
import org.apache.inlong.manager.dao.entity.InlongClusterNodeEntity;
+import org.apache.inlong.manager.dao.mapper.ComponentHeartbeatEntityMapper;
import org.apache.inlong.manager.dao.mapper.InlongClusterEntityMapper;
import org.apache.inlong.manager.dao.mapper.InlongClusterNodeEntityMapper;
import org.apache.inlong.manager.dao.mapper.StreamSourceEntityMapper;
@@ -87,6 +89,8 @@ public class HeartbeatManager implements
AbstractHeartbeatManager {
private InlongClusterNodeEntityMapper clusterNodeMapper;
@Autowired
private StreamSourceEntityMapper sourceMapper;
+ @Autowired
+ private ComponentHeartbeatEntityMapper componentHeartbeatMapper;
@Value("${cluster.heartbeat.interval:30}")
private Long heartbeatIntervalFactor;
@@ -223,6 +227,15 @@ public class HeartbeatManager implements
AbstractHeartbeatManager {
protocolTypes = null;
}
}
+ // If the manager has multiple nodes, need to determine that the
heartbeat is updated
+ // heartbeatInterval() is the reporting interval of cluster nodes,
multiplied by two to prevent network
+ // fluctuations
+ ComponentHeartbeatEntity componentHeartbeatEntity =
componentHeartbeatMapper.selectTimeOutHeartBeat(
+ componentHeartbeat.getComponentType(),
componentHeartbeat.getIp(), heartbeatInterval() * 2L);
+ if (componentHeartbeatEntity != null) {
+ heartbeatCache.put(componentHeartbeat, heartbeat);
+ return;
+ }
for (int i = 0; i < ports.length; i++) {
// deep clone the heartbeat
diff --git
a/inlong-manager/manager-web/src/main/resources/application-dev.properties
b/inlong-manager/manager-web/src/main/resources/application-dev.properties
index ae25f4da0..5f35ddc8f 100644
--- a/inlong-manager/manager-web/src/main/resources/application-dev.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-dev.properties
@@ -87,7 +87,7 @@ data.cleansing.batchSize=100
sort.enable.zookeeper=false
# cluster node timeout interval of heartbeat unit: second, the interval
multiplied by 5 represents the true heartbeat timeout interval
-cluster.heartbeat.interval=30
+cluster.heartbeat.interval=6
# If turned on, synchronizing change the source status when the agent
heartbeat times out
source.update.enabled=false
diff --git
a/inlong-manager/manager-web/src/main/resources/application-prod.properties
b/inlong-manager/manager-web/src/main/resources/application-prod.properties
index 7704c623c..62245389c 100644
--- a/inlong-manager/manager-web/src/main/resources/application-prod.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-prod.properties
@@ -86,7 +86,7 @@ data.cleansing.batchSize=100
sort.enable.zookeeper=false
# cluster node timeout interval of heartbeat unit: second, the interval
multiplied by 5 represents the true heartbeat timeout interval
-cluster.heartbeat.interval=30
+cluster.heartbeat.interval=6
# If turned on, synchronizing change the source status when the agent
heartbeat times out
source.update.enabled=false
diff --git
a/inlong-manager/manager-web/src/main/resources/application-test.properties
b/inlong-manager/manager-web/src/main/resources/application-test.properties
index 0988bf6a0..e0c0d8842 100644
--- a/inlong-manager/manager-web/src/main/resources/application-test.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-test.properties
@@ -87,7 +87,7 @@ data.cleansing.batchSize=100
sort.enable.zookeeper=false
# cluster node timeout interval of heartbeat unit: second, the interval
multiplied by 5 represents the true heartbeat timeout interval
-cluster.heartbeat.interval=30
+cluster.heartbeat.interval=6
# If turned on, synchronizing change the source status when the agent
heartbeat times out
source.update.enabled=false
