@SneakyThrows
    @Override
    public void reportHeartbeat(HeartbeatMsg heartbeat) {
        ComponentHeartbeat componentHeartbeat = heartbeat.componentHeartbeat();
        ClusterInfo clusterInfo = clusterInfoCache.get(componentHeartbeat);
        if (clusterInfo == null) {
            log.error("not found any cluster by name={} and type={}", componentHeartbeat.getClusterName(),
                    componentHeartbeat.getComponentType());
            return;
        }

        // if the heartbeat was not in the cache, insert or update the node by the heartbeat info
        HeartbeatMsg lastHeartbeat = heartbeatCache.getIfPresent(componentHeartbeat);
        if (heartbeat.getAddressInfos() != null) {
            heartbeat.setPort(Joiner.on(InlongConstants.COMMA)
                    .join(heartbeat.getAddressInfos().stream().map(AddressInfo::getPort).collect(Collectors.toList())));
            heartbeat.setIp(Joiner.on(InlongConstants.COMMA)
                    .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
        String[] ips = heartbeat.getIp().split(InlongConstants.COMMA);
        String port = heartbeat.getPort();
        String[] ports = null;
        if (StringUtils.isNotBlank(port)) {
            ports = port.split(InlongConstants.COMMA);
            if (ports.length < ips.length) {
                ports = null;
            }
        }
        String[] reportSourceTypes = null;
        if (StringUtils.isNotBlank(heartbeat.getReportSourceType()) && ips.length > 1) {
            reportSourceTypes = heartbeat.getReportSourceType().split(InlongConstants.COMMA);
            if (reportSourceTypes.length < ips.length) {
                reportSourceTypes = null;
            }
        }
        String protocolType = heartbeat.getProtocolType();
        String[] protocolTypes = null;
        if (StringUtils.isNotBlank(protocolType) && ips.length > 1) {
            protocolTypes = protocolType.split(InlongConstants.COMMA);
            if (protocolTypes.length < ips.length) {
                protocolTypes = null;
            }
        }

        int handlerNum = 0;
        for (int i = 0; i < ips.length; i++) {
            // deep clone the heartbeat
            HeartbeatMsg heartbeatMsg = JsonUtils.parseObject(JsonUtils.toJsonByte(heartbeat), HeartbeatMsg.class);
            assert heartbeatMsg != null;
            if (StringUtils.isNotBlank(port)) {
                heartbeatMsg.setPort(ports[i].trim());
            }
            heartbeatMsg.setIp(ips[i].trim());
            if (reportSourceTypes != null) {
                heartbeatMsg.setReportSourceType(reportSourceTypes[i].trim());
            } else {
                heartbeatMsg.setReportSourceType(heartbeat.getReportSourceType());
            }
            if (protocolTypes != null) {
                heartbeatMsg.setProtocolType(protocolTypes[i]);
            } else {
                heartbeatMsg.setProtocolType(protocolType);
            }
            if (Objects.equals(heartbeat.getComponentType(), ComponentTypeEnum.Agent.getType())) {
                heartbeatMsg.setProtocolType(null);
                heartbeatMsg.setPort(null);
            }
            // uninstall node event
            if (NodeSrvStatus.SERVICE_UNINSTALL.equals(heartbeat.getNodeSrvStatus())) {
                InlongClusterNodeEntity clusterNode = getClusterNode(clusterInfo, heartbeatMsg);
                deleteClusterNode(clusterNode);
                continue;
            }

            if (heartbeatConfigModified(lastHeartbeat, heartbeat)) {
                InlongClusterNodeEntity clusterNode = getClusterNode(clusterInfo, heartbeatMsg);
                if (clusterNode == null) {
                    handlerNum += insertClusterNode(clusterInfo, heartbeatMsg, clusterInfo.getCreator());
                } else {
                    handlerNum += updateClusterNode(clusterNode, heartbeatMsg);
                    // If the agent report succeeds, restore the source status
                    if (Objects.equals(clusterNode.getType(), ClusterType.AGENT)) {
                        // If the agent report succeeds, restore the source status
                        List<Integer> needUpdateIds = sourceMapper.selectHeartbeatTimeoutIds(null, heartbeat.getIp(),
                                heartbeat.getClusterName());
                        // restore state for all source by ip and type
                        if (CollectionUtils.isNotEmpty(needUpdateIds)) {
                            sourceMapper.rollbackTimeoutStatusByIds(needUpdateIds, null);
                        }
                    }
                }
            }
        }

        // if the heartbeat already exists, or does not exist but insert/update success, then put it into the cache
        if (lastHeartbeat == null || handlerNum == ips.length) {
            heartbeatCache.put(componentHeartbeat, heartbeat);
        }
    }
Check whether the configuration information carried in the heartbeat has been updated

@param oldHB last heartbeat msg
@param newHB current heartbeat msg
@return
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java