class RocketmqConsumer implements Closeable {
    static Logger log = LoggerFactory.getLogger(RocketmqConsumer.class);

    private RocketmqConfig config;

    ClientServiceProvider serviceProvider;
    private PushConsumer consumer;
    private RocketmqConsumerHandler handler;

    public RocketmqConsumer(RocketmqConfig config) {
        this.config = config;
    }

    public void init(CloudProps cloudProps, CloudEventObserverManger observerManger) throws ClientException {<FILL_FUNCTION_BODY>}

    @Override
    public void close() throws IOException {
        if(consumer != null){
            consumer.close();
        }
    }
}

        if (consumer != null) {
            return;
        }

        Utils.locker().lock();

        try {
            if (consumer != null) {
                return;
            }

            serviceProvider = ClientServiceProvider.loadService();

            ClientConfigurationBuilder builder = ClientConfiguration.newBuilder();

            //服务地址
            builder.setEndpoints(config.getServer());
            //账号密码
            if (Utils.isNotEmpty(config.getAccessKey())) {
                builder.setCredentialProvider(new StaticSessionCredentialsProvider(config.getAccessKey(), config.getSecretKey()));
            }

            //发送超时时间，默认3000 单位ms
            if (config.getTimeout() > 0) {
                builder.setRequestTimeout(Duration.ofMillis(config.getTimeout()));
            }

            ClientConfiguration configuration = builder.build();


            Map<String, FilterExpression> subscriptionExpressions = new LinkedHashMap<>();
            //要消费的topic，可使用tag进行简单过滤
            for (Map.Entry<String, Set<String>> kv : observerManger.topicTags().entrySet()) {
                String topic = kv.getKey();
                Set<String> tags = kv.getValue();
                String tagsExpr = String.join("||", tags);

                //支持 tag 过滤
                if (tags.contains("*")) {
                    subscriptionExpressions.put(topic, FilterExpression.SUB_ALL);
                } else {
                    subscriptionExpressions.put(topic, new FilterExpression(tagsExpr));
                }

                log.trace("Rocketmq5 consumer subscribe [" + topic + "(" + tagsExpr + ")] ok!");
            }

            PushConsumerBuilder consumerBuilder = serviceProvider.newPushConsumerBuilder();

            consumerBuilder.setClientConfiguration(configuration);
            //消费组
            consumerBuilder.setConsumerGroup(config.getConsumerGroup());
            //监听
            handler = new RocketmqConsumerHandler(config, observerManger);
            consumerBuilder.setMessageListener(handler);
            //订阅
            if (subscriptionExpressions.size() > 0) {
                consumerBuilder.setSubscriptionExpressions(subscriptionExpressions);
            }

            if (config.getConsumeThreadNums() > 0) {
                consumerBuilder.setConsumptionThreadCount(config.getConsumeThreadNums());
            }

            consumer = consumerBuilder.build();

            log.trace("Rocketmq5 consumer started!");
        } finally {
            Utils.locker().unlock();
        }
    
<no_super_class>