@Override
    public ListenerResult listen(WorkflowContext context) throws WorkflowListenerException {
        // query of inlong consume info from DB
        ApplyConsumeProcessForm consumeForm = (ApplyConsumeProcessForm) context.getProcessForm();
        Integer consumeId = consumeForm.getConsumeInfo().getId();
        InlongConsumeEntity entity = consumeMapper.selectById(consumeId);
        if (entity == null) {
            throw new WorkflowListenerException("inlong consume not exits for id=" + consumeId);
        }

        String mqType = entity.getMqType();
        if (MQType.TUBEMQ.equals(mqType)) {
            this.createTubeConsumerGroup(entity, context.getOperator());
        } else if (MQType.PULSAR.equals(mqType) || MQType.TDMQ_PULSAR.equals(mqType)) {
            this.createPulsarSubscription(entity);
        } else if (MQType.KAFKA.equals(mqType)) {
            // Kafka consumers do not need to register in advance
        } else {
            throw new WorkflowListenerException("Unsupported MQ type " + mqType);
        }

        // the consumer group may be changed when approving
        this.updateConsumerInfo(consumeId, entity.getConsumerGroup());
        return ListenerResult.success("Create MQ consumer group successful");
    }
Inlong consume process complete archive event listener
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/consume/apply/ApproveConsumeProcessListener.java