[Optimization][dinky-admin] Check if the cluster is disabled before s…

…tarting the flink session. (#4270)

Co-authored-by: yuhang2.zhang <yuhang2.zhang@ly.com>
Co-authored-by: GH Action - Upstream Sync <action@github.com>

dinky-admin/src/main/java/org/dinky/service/impl/ClusterConfigurationServiceImpl.java

@@ -75,7 +75,7 @@ public List<ClusterConfiguration> listAllClusterConfig() {
     public FlinkClusterConfig getAndCheckEnableFlinkClusterCfg(Integer id) {
         ClusterConfiguration cfg = this.getClusterConfigById(id);
         DinkyAssert.checkNull(cfg, "The clusterConfiguration not exists!");
-        DinkyAssert.checkEnable(cfg, "The clusterConfiguration is Disable!");
+        DinkyAssert.checkEnable(cfg, "The cluster is Disable!");
         return FlinkClusterConfig.create(cfg.getType(), cfg.getConfigJson());
     }
 

dinky-admin/src/main/java/org/dinky/service/impl/ClusterInstanceServiceImpl.java

@@ -34,7 +34,6 @@
 import org.dinky.data.model.Task;
 import org.dinky.gateway.config.FlinkConfig;
 import org.dinky.gateway.config.GatewayConfig;
-import org.dinky.gateway.exception.GatewayException;
 import org.dinky.gateway.model.FlinkClusterConfig;
 import org.dinky.gateway.result.GatewayResult;
 import org.dinky.job.JobConfig;
@@ -230,9 +229,8 @@ public void killCluster(Integer id) {
     @Override
     public ClusterInstance deploySessionCluster(Integer id) {
         ClusterConfiguration clusterCfg = clusterConfigurationService.getClusterConfigById(id);
-        if (Asserts.isNull(clusterCfg)) {
-            throw new GatewayException("The cluster configuration does not exist.");
-        }
+        DinkyAssert.checkNull(clusterCfg, "The clusterConfiguration not exists!");
+        DinkyAssert.checkEnable(clusterCfg, "The cluster is Disable!");
 
         // add custom configuration.
         FlinkConfig flinkConfig = clusterCfg.getConfigJson().getFlinkConfig();