private void sendResponse(ChannelHandlerContext ctx,
            int errCode, String errMsg, boolean isClose, String callback) {
        if (ctx == null || ctx.channel() == null) {
            return;
        }
        if (!ctx.channel().isWritable()) {
            source.fileMetricIncSumStats(StatConstants.EVENT_HTTP_LINK_UNWRITABLE);
            if (logCounter.shouldPrint()) {
                logger.warn("Send msg but channel full, channel={}", ctx.channel());
            }
            return;
        }
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpAttrConst.RET_CNT_TYPE);
        StringBuilder builder = new StringBuilder(512);
        if (StringUtils.isNotBlank(callback)) {
            builder.append(callback).append("(");
        }
        builder.append("{\"code\":\"").append(errCode)
                .append("\",\"msg\":\"").append(errMsg).append("\"}");
        if (StringUtils.isNotBlank(callback)) {
            builder.append(")");
        }
        ByteBuf buffer = Unpooled.copiedBuffer(builder.toString(), CharsetUtil.UTF_8);
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, buffer.readableBytes());
        response.content().writeBytes(buffer);
        buffer.release();
        ctx.writeAndFlush(response).addListener(new SendResultListener(isClose));
    }
Get attributes from decoder

@param decoder the decode object
@param reqAttrs the attributes
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/source/httpMsg/HttpMessageHandler.java