private void procBinHeartbeatMsg(BaseSource source, Channel channel,
            ByteBuf cb, int totalDataLen) throws Exception {
        // Check if the message is complete and legal
        if (totalDataLen < BIN_HB_FIXED_CONTENT_SIZE) {
            source.fileMetricIncSumStats(StatConstants.EVENT_MSG_HB_TOTALLEN_BELOWMIN);
            String errMsg = String.format("Malformed msg, totalDataLen(%d) < min hb-msg length(%d)",
                    totalDataLen, BIN_HB_FIXED_CONTENT_SIZE);
            if (logger.isDebugEnabled()) {
                logger.debug(errMsg + ", channel {}", channel);
            }
            throw new PkgParseException(errMsg);
        }
        // check validation
        int msgHeadPos = cb.readerIndex() - 5;
        int bodyLen = cb.getInt(msgHeadPos + BIN_HB_BODYLEN_OFFSET);
        int attrLen = cb.getShort(msgHeadPos + BIN_HB_BODY_OFFSET + bodyLen);
        int msgMagic = cb.getUnsignedShort(msgHeadPos
                + BIN_HB_BODY_OFFSET + bodyLen + BIN_HB_ATTRLEN_SIZE + attrLen);
        if (msgMagic != BIN_MSG_MAGIC) {
            source.fileMetricIncSumStats(StatConstants.EVENT_MSG_HB_MAGIC_UNEQUAL);
            String errMsg = String.format(
                    "Malformed msg, msgMagic(%d) != %d", msgMagic, BIN_MSG_MAGIC);
            if (logger.isDebugEnabled()) {
                logger.debug(errMsg + ", channel {}", channel);
            }
            throw new PkgParseException(errMsg);
        }
        if (totalDataLen + BIN_HB_TOTALLEN_SIZE < (bodyLen + attrLen + BIN_HB_FORMAT_SIZE)) {
            source.fileMetricIncSumStats(StatConstants.EVENT_MSG_HB_LEN_MALFORMED);
            String errMsg = String.format(
                    "Malformed msg, bodyLen(%d) + attrLen(%d) > totalDataLen(%d)",
                    bodyLen, attrLen, totalDataLen);
            if (logger.isDebugEnabled()) {
                logger.debug(errMsg + ", channel {}", channel);
            }
            throw new PkgParseException(errMsg);
        }
        // read message content
        byte version = cb.getByte(msgHeadPos + BIN_HB_VERSION_OFFSET);
        byte[] attrData = null;
        if (attrLen > 0) {
            attrData = new byte[attrLen];
            cb.getBytes(msgHeadPos + BIN_HB_BODY_OFFSET
                    + bodyLen + BIN_HB_ATTRLEN_SIZE, attrData, 0, attrLen);
        }
        // build and send response message
        flushV0MsgPackage(source, channel,
                buildHBRspPackage(attrData, version, 0), MsgType.MSG_BIN_HEARTBEAT.name());
    }
extract and process bin heart beat msg, message type is 8
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/ServerMessageHandler.java