@Override
    public void deleteQueueForGroup(InlongGroupInfo groupInfo, String operator) {
        Preconditions.expectNotNull(groupInfo, "inlong group info cannot be null");
        String groupId = groupInfo.getInlongGroupId();
        String clusterTag = groupInfo.getInlongClusterTag();
        log.info("begin to delete pulsar resource for groupId={}, clusterTag={}", groupId, clusterTag);

        List<InlongStreamBriefInfo> streamInfos = streamService.getTopicList(groupId);
        if (CollectionUtils.isEmpty(streamInfos)) {
            log.warn("skip to delete pulsar resource as no streams for groupId={}", groupId);
            return;
        }

        List<ClusterInfo> clusterInfos = clusterService.listByTagAndType(clusterTag, ClusterType.PULSAR);
        for (ClusterInfo clusterInfo : clusterInfos) {
            PulsarClusterInfo pulsarCluster = (PulsarClusterInfo) clusterInfo;
            try {
                for (InlongStreamBriefInfo streamInfo : streamInfos) {
                    this.deletePulsarTopic((InlongPulsarInfo) groupInfo, pulsarCluster, streamInfo.getMqResource());
                }
            } catch (Exception e) {
                String msg = "failed to delete pulsar resource for groupId=" + groupId;
                log.error(msg + ", cluster=" + pulsarCluster, e);
                throw new WorkflowListenerException(msg + ": " + e.getMessage());
            }
        }

        log.info("success to delete pulsar resource for groupId={}, clusterTag={}", groupId, clusterTag);
    }
The name rule for Pulsar subscription: clusterTag_topicName_sinkId_consumer_group
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarQueueResourceOperator.java