class OnsConsumer {
    static Logger log = LoggerFactory.getLogger(OnsConsumer.class);

    private OnsConfig config;
    private ConsumerBean consumer;

    public OnsConsumer(OnsConfig config) {
        this.config = config;
    }

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

        if (consumer != null) {
            return;
        }

        Utils.locker().lock();

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

            consumer = new ConsumerBean();
            consumer.setProperties(config.getConsumerProperties());

            OnsConsumerHandler handler = new OnsConsumerHandler(config, observerManger);

            Map<Subscription, MessageListener> subscriptionTable = new HashMap<>();
            for (Map.Entry<String, Set<String>> kv : observerManger.topicTags().entrySet()) {
                String topic = kv.getKey();
                Set<String> tags = kv.getValue();
                String tagsExpr = String.join("||", tags);

                Subscription subscription = new Subscription();
                subscription.setTopic(topic);
                if (tags.contains("*")) {
                    subscription.setExpression("*");
                } else {
                    subscription.setExpression(tagsExpr);
                }

                subscriptionTable.put(subscription, handler);

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

            consumer.setSubscriptionTable(subscriptionTable);
            consumer.start();

            if (consumer.isStarted()) {
                log.trace("Ons consumer started!");
            } else {
                log.warn("Ons consumer start failure!");
            }
        } finally {
            Utils.locker().unlock();
        }
    
<no_super_class>