[INLONG-6352][Manager] Add a global config for whether to use ZooKeep…

…er to manage the Sort task (#6354)

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java

@@ -64,6 +64,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Isolation;
 import org.springframework.transaction.annotation.Propagation;
@@ -91,6 +92,9 @@ public class InlongGroupServiceImpl implements InlongGroupService {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(InlongGroupServiceImpl.class);
 
+    @Value("${sort.enable.zookeeper:false}")
+    private boolean enableZookeeper;
+
     @Autowired
     private InlongGroupEntityMapper groupMapper;
     @Autowired
@@ -154,7 +158,7 @@ public String save(InlongGroupRequest request, String operator) {
             LOGGER.error("groupId {} has already exists", groupId);
             throw new BusinessException(ErrorCodeEnum.GROUP_DUPLICATE);
         }
-
+        request.setEnableZookeeper(enableZookeeper ? InlongConstants.ENABLE_ZK : InlongConstants.DISABLE_ZK);
         InlongGroupOperator instance = groupOperatorFactory.getInstance(request.getMqType());
         groupId = instance.saveOpt(request, operator);
 
@@ -246,6 +250,7 @@ public String update(InlongGroupRequest request, String operator) {
         // check whether the current status can be modified
         checkGroupCanUpdate(entity, request, operator);
 
+        request.setEnableZookeeper(enableZookeeper ? InlongConstants.ENABLE_ZK : InlongConstants.DISABLE_ZK);
         InlongGroupOperator instance = groupOperatorFactory.getInstance(request.getMqType());
         instance.updateOpt(request, operator);
 

inlong-manager/manager-web/src/main/resources/application-dev.properties

@@ -82,3 +82,6 @@ data.cleansing.interval.seconds=1800
 data.cleansing.before.days=10
 # The maximum size of data to be deleted in batch, default is 100
 data.cleansing.batchSize=100
+
+# Whether to use ZooKeeper to manage the Sort task config, default is false, which means not using ZooKeeper
+sort.enable.zookeeper=false

inlong-manager/manager-web/src/main/resources/application-prod.properties

@@ -80,3 +80,6 @@ data.cleansing.interval.seconds=1800
 data.cleansing.before.days=10
 # The maximum size of data to be deleted in batch, default is 100
 data.cleansing.batchSize=100
+
+# Whether to use ZooKeeper to manage the Sort task config, default is false, which means not using ZooKeeper
+sort.enable.zookeeper=false

inlong-manager/manager-web/src/main/resources/application-test.properties

@@ -81,3 +81,6 @@ data.cleansing.interval.seconds=1800
 data.cleansing.before.days=10
 # The maximum size of data to be deleted in batch, default is 100
 data.cleansing.batchSize=100
+
+# Whether to use ZooKeeper to manage the Sort task config, default is false, which means not using ZooKeeper
+sort.enable.zookeeper=false