public static void sinkReturnRspPackage(SinkRspEvent event,
            DataProxyErrCode errCode,
            String errMsg) {
        ByteBuf binBuffer;
        final StringBuilder strBuff = new StringBuilder(512);
        // get and check channel context
        Channel remoteChannel = event.getChannel();
        if (remoteChannel == null || !remoteChannel.isActive()) {
            return;
        }
        // check message type
        MsgType msgType = event.getMsgType();
        if (MsgType.MSG_UNKNOWN.equals(msgType)
                || MsgType.MSG_ACK_SERVICE.equals(msgType)
                || MsgType.MSG_HEARTBEAT.equals(msgType)
                || MsgType.MSG_BIN_HEARTBEAT.equals(msgType)) {
            return;
        }
        // check whether return response message
        Map<String, String> attrMap = event.getHeaders();
        String isAck = attrMap.get(AttributeConstants.MESSAGE_IS_ACK);
        if ("false".equalsIgnoreCase(isAck)) {
            return;
        }
        String origAttrs = attrMap.get(ConfigConstants.DECODER_ATTRS);
        // check whether channel is writable
        if (!remoteChannel.isWritable()) {
            if (logCounter.shouldPrint()) {
                logger.warn(strBuff.append("Send buffer is full3 by channel ")
                        .append(remoteChannel).append(", attr is ").append(origAttrs).toString());
            }
            return;
        }
        // build return attribute string
        strBuff.append(ConfigConstants.DATAPROXY_IP_KEY)
                .append(AttributeConstants.KEY_VALUE_SEPARATOR).append(getLocalIp());
        if (DataProxyErrCode.SUCCESS != errCode) {
            strBuff.append(AttributeConstants.SEPARATOR)
                    .append(AttributeConstants.MESSAGE_PROCESS_ERRCODE)
                    .append(AttributeConstants.KEY_VALUE_SEPARATOR).append(errCode.getErrCode());
            if (StringUtils.isNotEmpty(errMsg)) {
                strBuff.append(AttributeConstants.SEPARATOR)
                        .append(AttributeConstants.MESSAGE_PROCESS_ERRMSG)
                        .append(AttributeConstants.KEY_VALUE_SEPARATOR).append(errMsg);
            }
        }
        if (StringUtils.isNotEmpty(origAttrs)) {
            strBuff.append(AttributeConstants.SEPARATOR).append(origAttrs);
        }
        String destAttrs = strBuff.toString();
        // build response message bytes
        if (MsgType.MSG_BIN_MULTI_BODY.equals(msgType)) {
            binBuffer = buildBinMsgRspPackage(destAttrs,
                    attrMap.get(AttributeConstants.UNIQ_ID));
        } else {
            // MsgType.MSG_ORIGINAL_RETURN.equals(msgType)
            // MsgType.MSG_MULTI_BODY.equals(msgType)
            // MsgType.MSG_MULTI_BODY_ATTR.equals(msgType)
            binBuffer = buildDefMsgRspPackage(msgType, destAttrs);
        }
        // send response message
        try {
            if (remoteChannel.isWritable()) {
                remoteChannel.writeAndFlush(binBuffer);
            } else {
                // release allocated ByteBuf
                binBuffer.release();
                if (logCounter.shouldPrint()) {
                    strBuff.delete(0, strBuff.length());
                    logger.warn(strBuff.append("Send buffer is full4 by channel ")
                            .append(remoteChannel).append(", attr is ").append(origAttrs).toString());
                }
            }
        } catch (Throwable e) {
            // should release bin buffer?
            if (logCounter.shouldPrint()) {
                strBuff.delete(0, strBuff.length());
                logger.error(strBuff.append("Write data to channel exception, channel=")
                        .append(remoteChannel).append(", attr is ").append(origAttrs).toString());
            }
        }
    }
Return response to client in sink
@param event    the event need to response
@param errCode  process error code
@param errMsg   error message
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/MessageUtils.java