@Override
        public void channelRead(ChannelHandlerContext ctx, Object e) {
            if (e instanceof RpcDataPack) {
                RpcDataPack dataPack = (RpcDataPack) e;
                Callback callback = requests.remove(dataPack.getSerialNo());
                if (callback != null) {
                    Timeout timeout = timeouts.remove(dataPack.getSerialNo());
                    if (timeout != null) {
                        timeout.cancel();
                    }
                    ResponseWrapper responseWrapper;
                    try {
                        ByteBufferInputStream in = new ByteBufferInputStream(dataPack.getDataLst());
                        RPCProtos.RpcConnHeader connHeader =
                                RPCProtos.RpcConnHeader.parseDelimitedFrom(in);
                        if (connHeader == null) {
                            // When the stream is closed, protobuf doesn't raise an EOFException,
                            // instead, it returns a null message object.
                            throw new EOFException();
                        }
                        RPCProtos.ResponseHeader rpcResponse =
                                RPCProtos.ResponseHeader.parseDelimitedFrom(in);
                        if (rpcResponse == null) {
                            // When the stream is closed, protobuf doesn't raise an EOFException,
                            // instead, it returns a null message object.
                            throw new EOFException();
                        }
                        RPCProtos.ResponseHeader.Status status = rpcResponse.getStatus();
                        if (status == RPCProtos.ResponseHeader.Status.SUCCESS) {
                            RPCProtos.RspResponseBody pbRpcResponse =
                                    RPCProtos.RspResponseBody.parseDelimitedFrom(in);
                            if (pbRpcResponse == null) {
                                // When the RPCProtos parse failed , protobuf doesn't raise an Exception,
                                // instead, it returns a null response object.
                                throw new NetworkException("Not found PBRpcResponse data!");
                            }
                            Object responseResult =
                                    PbEnDecoder.pbDecode(false, pbRpcResponse.getMethod(),
                                            pbRpcResponse.getData().toByteArray());

                            responseWrapper =
                                    new ResponseWrapper(connHeader.getFlag(), dataPack.getSerialNo(),
                                            rpcResponse.getServiceType(), rpcResponse.getProtocolVer(),
                                            pbRpcResponse.getMethod(), responseResult);
                        } else {
                            RPCProtos.RspExceptionBody exceptionResponse =
                                    RPCProtos.RspExceptionBody.parseDelimitedFrom(in);
                            if (exceptionResponse == null) {
                                // When the RPCProtos parse failed , protobuf doesn't raise an Exception,
                                // instead, it returns a null response object.
                                throw new NetworkException("Not found RpcException data!");
                            }
                            String exceptionName = exceptionResponse.getExceptionName();
                            exceptionName = MixUtils.replaceClassNamePrefix(exceptionName,
                                    false, rpcResponse.getProtocolVer());
                            responseWrapper =
                                    new ResponseWrapper(connHeader.getFlag(), dataPack.getSerialNo(),
                                            rpcResponse.getServiceType(), rpcResponse.getProtocolVer(),
                                            exceptionName, exceptionResponse.getStackTrace());
                        }
                        if (!responseWrapper.isSuccess()) {
                            Throwable remote =
                                    MixUtils.unwrapException(new StringBuilder(512)
                                            .append(responseWrapper.getErrMsg()).append("#")
                                            .append(responseWrapper.getStackTrace()).toString());
                            if (IOException.class.isAssignableFrom(remote.getClass())) {
                                NettyClient.this.close();
                            }
                        }
                        callback.handleResult(responseWrapper);
                    } catch (Throwable ee) {
                        responseWrapper =
                                new ResponseWrapper(-2, dataPack.getSerialNo(), -2, -2, -2, ee);
                        if (ee instanceof EOFException) {
                            NettyClient.this.close();
                        }
                        callback.handleResult(responseWrapper);
                    }
                } else {
                    if (logger.isDebugEnabled()) {
                        logger.debug("Missing previous call info, maybe it has been timeout.");
                    }
                }
            }
        }
Invoked when a message object was received from a remote peer.

@param ctx     the channel handler context
@param e       the message event
inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClient.java