public static Map<String, String> getXfsAttrs(Map<String, String> headers, String pkgVersion) {
        // common attributes
        Map<String, String> attrs = new HashMap<>();
        attrs.put(ConfigConstants.MSG_ENCODE_VER, pkgVersion);
        if (MessageWrapType.INLONG_MSG_V1.getStrId().equalsIgnoreCase(pkgVersion)) {
            attrs.put("dataproxyip", NetworkUtils.getLocalIp());
            attrs.put(Constants.INLONG_GROUP_ID, headers.get(Constants.INLONG_GROUP_ID));
            attrs.put(Constants.INLONG_STREAM_ID, headers.get(Constants.INLONG_STREAM_ID));
            attrs.put(Constants.TOPIC, headers.get(Constants.TOPIC));
            attrs.put(Constants.HEADER_KEY_MSG_TIME, headers.get(Constants.HEADER_KEY_MSG_TIME));
            attrs.put(Constants.HEADER_KEY_SOURCE_IP, headers.get(Constants.HEADER_KEY_SOURCE_IP));
        } else {
            //
            attrs.put(Constants.INLONG_GROUP_ID,
                    headers.get(AttributeConstants.GROUP_ID));
            attrs.put(Constants.INLONG_STREAM_ID,
                    headers.get(AttributeConstants.STREAM_ID));
            attrs.put(Constants.TOPIC,
                    headers.get(ConfigConstants.TOPIC_KEY));
            attrs.put(Constants.HEADER_KEY_MSG_TIME,
                    headers.get(AttributeConstants.DATA_TIME));
            attrs.put(Constants.HEADER_KEY_SOURCE_IP,
                    headers.get(ConfigConstants.REMOTE_IP_KEY));
            attrs.put(Constants.HEADER_KEY_SOURCE_TIME,
                    headers.get(AttributeConstants.RCV_TIME));
            attrs.put(ConfigConstants.DATAPROXY_IP_KEY,
                    NetworkUtils.getLocalIp());
        }
        return attrs;
    }
Build heartbeat response message ByteBuf

@param attrs     the attribute string
@param version   the version
@param loadValue the node load value
@return ByteBuf
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java