[INLONG-9968][Manager] Support pulsar multi cluster when creating pul…

…sar consumption groups (apache#9970)
XiaoYou201 · Apr 10, 2024 · bff9c2c · bff9c2c
1 parent dba5900
commit bff9c2c
Showing 1 changed file with 21 additions and 17 deletions.
diff --git a/...g/apache/inlong/manager/service/listener/consume/apply/ApproveConsumeProcessListener.java b/...g/apache/inlong/manager/service/listener/consume/apply/ApproveConsumeProcessListener.java
@@ -127,24 +127,28 @@ private void createPulsarSubscription(InlongConsumeEntity entity) {
  "mq resource cannot empty for groupId=" + groupId);
 
  String clusterTag = groupEntity.getInlongClusterTag();
- ClusterInfo clusterInfo = clusterService.getOne(clusterTag, null, ClusterType.PULSAR);
- PulsarClusterInfo pulsarCluster = (PulsarClusterInfo) clusterInfo;
- try {
- InlongPulsarDTO pulsarDTO = InlongPulsarDTO.getFromJson(groupEntity.getExtParams());
- String tenant = pulsarDTO.getPulsarTenant();
- if (StringUtils.isBlank(tenant)) {
- tenant = pulsarCluster.getPulsarTenant();
+ List<ClusterInfo> clusterInfos = clusterService.listByTagAndType(clusterTag, ClusterType.PULSAR);
+ for (ClusterInfo clusterInfo : clusterInfos) {
+ PulsarClusterInfo pulsarCluster = (PulsarClusterInfo) clusterInfo;
+ try {
+ InlongPulsarDTO pulsarDTO = InlongPulsarDTO.getFromJson(groupEntity.getExtParams());
+ String tenant = pulsarDTO.getPulsarTenant();
+ if (StringUtils.isBlank(tenant)) {
+ tenant = pulsarCluster.getPulsarTenant();
+ }
+ PulsarTopicInfo topicMessage = new PulsarTopicInfo();
+ topicMessage.setPulsarTenant(tenant);
+ topicMessage.setNamespace(mqResource);
+
+ List<String> topics = Arrays.asList(entity.getTopic().split(InlongConstants.COMMA));
+
+ this.createPulsarSubscription(pulsarCluster, entity.getConsumerGroup(), topicMessage, topics);
+ } catch (Exception e) {
+ log.error("create pulsar topic failed", e);
+ throw new WorkflowListenerException(
+ "failed to create pulsar topic for groupId=" + groupId + ", reason: "
+ + e.getMessage());
  }
- PulsarTopicInfo topicMessage = new PulsarTopicInfo();
- topicMessage.setPulsarTenant(tenant);
- topicMessage.setNamespace(mqResource);
-
- List<String> topics = Arrays.asList(entity.getTopic().split(InlongConstants.COMMA));
- this.createPulsarSubscription(pulsarCluster, entity.getConsumerGroup(), topicMessage, topics);
- } catch (Exception e) {
- log.error("create pulsar topic failed", e);
- throw new WorkflowListenerException("failed to create pulsar topic for groupId=" + groupId + ", reason: "
- + e.getMessage());
  }
  }