[INLONG-9484]][Manager] Improve logic of sortstandalone sink auto-ass…

…igned cluster (apache#9485)
featzhang · Dec 15, 2023 · 47c4fa0 · 47c4fa0
1 parent bfc7750
commit 47c4fa0
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 17 deletions.
diff --git a/...ger/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterRequest.java b/...ger/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/ClusterRequest.java
@@ -48,7 +48,7 @@ public abstract class ClusterRequest {
     private Integer id;
 
     @ApiModelProperty(value = "Cluster name")
-    @Pattern(regexp = "^[A-Za-z0-9_-]{1,128}$", message = "only supports letters, numbers, '-', or '_'")
+    @Pattern(regexp = "^[A-Za-z0-9._-]{1,128}$", message = "only supports letters, numbers, '.', '-', or '_'")
     private String name;
 
     @ApiModelProperty(value = "Cluster display name, just for display")

diff --git a/...nager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodeRequest.java b/...nager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/node/DataNodeRequest.java
@@ -48,7 +48,7 @@ public abstract class DataNodeRequest {
     private Integer id;
 
     @ApiModelProperty(value = "Data node name")
-    @Pattern(regexp = "^[A-Za-z0-9_-]{1,128}$", message = "only supports letters, numbers, '-', or '_'")
+    @Pattern(regexp = "^[A-Za-z0-9._-]{1,128}$", message = "only supports letters, numbers, '.', '-', or '_'")
     private String name;
 
     @ApiModelProperty(value = "Data node display name, just for display")

diff --git a/...g/apache/inlong/manager/service/resource/sink/AbstractStandaloneSinkResourceOperator.java b/...g/apache/inlong/manager/service/resource/sink/AbstractStandaloneSinkResourceOperator.java
@@ -17,8 +17,8 @@
 
 package org.apache.inlong.manager.service.resource.sink;
 
-import org.apache.inlong.manager.common.consts.InlongConstants;
 import org.apache.inlong.manager.common.consts.SinkType;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
 import org.apache.inlong.manager.common.enums.SinkStatus;
 import org.apache.inlong.manager.common.util.Preconditions;
 import org.apache.inlong.manager.dao.entity.InlongClusterEntity;
@@ -31,15 +31,15 @@
 import org.apache.inlong.manager.service.sink.StreamSinkService;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Sets;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.util.CollectionUtils;
 
 import java.util.List;
 import java.util.Random;
-import java.util.stream.Collectors;
 
+@Slf4j
 public abstract class AbstractStandaloneSinkResourceOperator implements SinkResourceOperator {
 
     @Autowired
@@ -51,14 +51,16 @@ public abstract class AbstractStandaloneSinkResourceOperator implements SinkReso
     @Autowired
     private StreamSinkService sinkService;
 
-    private static final String SORT_PREFIX = "SORT_";
-
     private Random rand = new Random();
 
     @VisibleForTesting
     protected void assignCluster(SinkInfo sinkInfo) {
+        if (StringUtils.isBlank(sinkInfo.getSinkType())) {
+            throw new IllegalArgumentException(ErrorCodeEnum.SINK_TYPE_IS_NULL.getMessage());
+        }
+
         if (StringUtils.isNotBlank(sinkInfo.getInlongClusterName())) {
-            String info = "success to create es resource";
+            String info = "no need to auto-assign cluster since the cluster has already assigned";
             sinkService.updateStatus(sinkInfo.getId(), SinkStatus.CONFIG_SUCCESSFUL.getCode(), info);
             return;
         }
@@ -88,18 +90,23 @@ private String assignFromExist(String dataNodeName) {
     private String assignFromRelated(String sinkType, String groupId) {
         InlongGroupEntity group = groupEntityMapper.selectByGroupId(groupId);
         String sortClusterType = SinkType.relatedSortClusterType(sinkType);
-        List<InlongClusterEntity> clusters = clusterEntityMapper
-                .selectByKey(null, null, sortClusterType).stream()
-                .filter(cluster -> checkCluster(cluster.getClusterTags(), group.getInlongClusterTag()))
-                .collect(Collectors.toList());
+        if (StringUtils.isBlank(sortClusterType)) {
+            log.error("find no relate sort cluster type for sink type={}", sinkType);
+            return null;
+        }
 
-        return CollectionUtils.isEmpty(clusters) ? null : clusters.get(rand.nextInt(clusters.size())).getName();
+        // if some clusters have the same tag
+        List<InlongClusterEntity> clusters =
+                clusterEntityMapper.selectByKey(group.getInlongClusterTag(), null, sortClusterType);
+        if (!CollectionUtils.isEmpty(clusters)) {
+            return clusters.get(rand.nextInt(clusters.size())).getName();
+        }
 
-    }
+        // if no cluster with the same tag
+        clusters = clusterEntityMapper.selectByKey(null, null, sortClusterType);
+
+        return CollectionUtils.isEmpty(clusters) ? null : clusters.get(rand.nextInt(clusters.size())).getName();
 
-    private boolean checkCluster(String clusterTags, String targetTag) {
-        return StringUtils.isBlank(clusterTags)
-                || Sets.newHashSet(clusterTags.split(InlongConstants.COMMA)).contains(targetTag);
     }
 
 }