private void responseV0Msg(DataProxyErrCode errCode, String errMsg) {
        String uniqId = event.getHeaders().get(AttributeConstants.UNIQ_ID);
        if ("false".equals(event.getHeaders().get(AttributeConstants.MESSAGE_IS_ACK))) {
            if (logger.isDebugEnabled()) {
                logger.debug("not need to rsp message: uniqId = {}, inlongGroupId = {}, inlongStreamId = {}",
                        uniqId, this.getInlongGroupId(), this.getInlongStreamId());
            }
            return;
        }
        // check channel status
        if (channel == null || !channel.isWritable()) {
            if (logCounter.shouldPrint()) {
                logger.warn("Prepare send msg but channel full, msgType={}, attr={}, channel={}",
                        msgType, event.getHeaders(), channel);
            }
            return;
        }
        try {
            // build return attribute string
            StringBuilder strBuff = new StringBuilder(512);
            if (errCode != DataProxyErrCode.SUCCESS) {
                strBuff.append(AttributeConstants.MESSAGE_PROCESS_ERRCODE)
                        .append(AttributeConstants.KEY_VALUE_SEPARATOR).append(errCode.getErrCodeStr());
                if (StringUtils.isNotEmpty(errMsg)) {
                    strBuff.append(AttributeConstants.SEPARATOR).append(AttributeConstants.MESSAGE_PROCESS_ERRMSG)
                            .append(AttributeConstants.KEY_VALUE_SEPARATOR).append(errMsg);
                }
            }
            String origAttr = event.getHeaders().getOrDefault(ConfigConstants.DECODER_ATTRS, "");
            if (StringUtils.isNotEmpty(origAttr)) {
                if (strBuff.length() > 0) {
                    strBuff.append(AttributeConstants.SEPARATOR).append(origAttr);
                } else {
                    strBuff.append(origAttr);
                }
            }
            // build and send response message
            ByteBuf retData;
            if (MsgType.MSG_BIN_MULTI_BODY.equals(msgType)) {
                retData = ServerMessageHandler.buildBinMsgRspPackage(strBuff.toString(), Long.parseLong(uniqId));
            } else {
                retData = ServerMessageHandler.buildTxtMsgRspPackage(msgType, strBuff.toString());
            }
            strBuff.delete(0, strBuff.length());
            if (channel == null || !channel.isWritable()) {
                // release allocated ByteBuf
                retData.release();
                if (logCounter.shouldPrint()) {
                    logger.warn("Send msg but channel full, attr={}, channel={}", event.getHeaders(), channel);
                }
                return;
            }
            channel.writeAndFlush(retData);
        } catch (Throwable e) {
            //
            if (logCounter.shouldPrint()) {
                logger.warn("Send msg but failure, attr={}", event.getHeaders(), e);
            }
        }
    }
Return response to client in source
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/sink/mq/SimplePackProfile.java