@Override
            public void run() {
                while (!isStopped) {
                    try {
                        long currentTime = System.currentTimeMillis();
                        Set<String> removedNodeKey = new HashSet<>();
                        for (Map.Entry<String, TimeoutInfo> entry : nodeMap.entrySet()) {
                            if (TStringUtils.isBlank(entry.getKey()) || entry.getValue() == null) {
                                continue;
                            }
                            if (currentTime >= entry.getValue().getTimeoutTime()) {
                                removedNodeKey.add(entry.getKey());
                            }
                        }
                        if (!removedNodeKey.isEmpty()) {
                            for (String nodeKey : removedNodeKey) {
                                TimeoutInfo timeoutInfo = nodeMap.get(nodeKey);
                                if (timeoutInfo == null) {
                                    continue;
                                }
                                if (currentTime >= timeoutInfo.getTimeoutTime()) {
                                    nodeMap.remove(nodeKey);
                                    listener.onTimeout(nodeKey, timeoutInfo);
                                }
                            }
                        }
                        Thread.sleep(1000 * 1);
                    } catch (Throwable t) {
                        logger.error(new StringBuilder(256)
                                .append(businessType).append(" heartbeat scan error!").toString(), t);
                    }
                }
            }
Register the check business for consumer.

@param timeout  the timeout to be registered for the consumer.
@param listener the listener used in the consumer for timeout business
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/common/heartbeat/HeartbeatManager.java