class RocketmqConsumerHandler implements MessageListener {
    static Logger log = LoggerFactory.getLogger(RocketmqConsumerHandler.class);

    private final CloudEventObserverManger observerManger;
    private final RocketmqConfig config;

    public RocketmqConsumerHandler(RocketmqConfig config, CloudEventObserverManger observerManger) {
        this.observerManger = observerManger;
        this.config = config;
    }


    @Override
    public ConsumeResult consume(MessageView message) {<FILL_FUNCTION_BODY>}

    /**
     * 处理接收事件
     */
    protected boolean onReceive(Event event, String topicNew) throws Throwable {
        boolean isOk = true;
        CloudEventHandler handler = null;

        if (Utils.isEmpty(event.tags())) {
            handler = observerManger.getByTopicAndTag(topicNew, "*");
        } else {
            handler = observerManger.getByTopicAndTag(topicNew, event.tags());

            if (handler == null) {
                handler = observerManger.getByTopicAndTag(topicNew, "*");
            }
        }

        if (handler != null) {
            isOk = handler.handle(event);
        } else {
            //只需要记录一下
            log.warn("There is no observer for this event topic[{}]", topicNew);
        }

        return isOk;
    }
}

        boolean isOk = true;

        try {
            String topicNew = message.getTopic();
            String group = null;
            String topic = null;
            if (topicNew.contains(RocketmqProps.GROUP_SPLIT_MARK)) {
                group = topicNew.split(RocketmqProps.GROUP_SPLIT_MARK)[0];
                topic = topicNew.split(RocketmqProps.GROUP_SPLIT_MARK)[1];
            } else {
                topic = topicNew;
            }

            String content = StandardCharsets.UTF_8.decode(message.getBody()).toString();

            Event event = new Event(topic, content);
            event.tags(message.getTag().orElse(null));
            event.key(String.join(",", message.getKeys()));
            event.times(message.getDeliveryAttempt() - 1); //它是从1开始的
            event.channel(config.getChannelName());
            if (Utils.isNotEmpty(group)) {
                event.group(group);
            }

            isOk = isOk && onReceive(event, topicNew); //可以不吃异常

        } catch (Throwable e) {
            isOk = false;
            log.warn(e.getMessage(), e);
        }

        if (isOk) {
            return ConsumeResult.SUCCESS;
        } else {
            return ConsumeResult.FAILURE;
        }
    
<no_super_class>