[Feature-2622][core] Local task supports cancel (DataLinkDC#2623)

Co-authored-by: wenmo <32723967+wenmo@users.noreply.github.com>
hugh018 · Dec 12, 2023 · 7ea8cab · 7ea8cab
1 parent 963cb1d
commit 7ea8cab
Show file tree

Hide file tree

Showing 6 changed files with 24 additions and 4 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/job/handler/Job2MysqlHandler.java b/dinky-admin/src/main/java/org/dinky/job/handler/Job2MysqlHandler.java
@@ -129,7 +129,7 @@ public boolean success() {
         history.setStatus(job.getStatus().getCode());
         history.setJobId(job.getJobId());
         history.setEndTime(job.getEndTime());
-        history.setJobManagerAddress(job.isUseGateway() ? job.getJobManagerAddress() : null);
+        history.setJobManagerAddress(job.getJobManagerAddress());
 
         Integer clusterId = job.getJobConfig().getClusterId();
         ClusterInstance clusterInstance;

diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/ClusterInstanceServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/ClusterInstanceServiceImpl.java
@@ -100,6 +100,9 @@ public String buildEnvironmentAddress(JobConfig config) {
                 port = Integer.valueOf(flinkConfig.get("rest.port"));
             } else {
                 port = URLUtils.getRandomPort();
+                while (!IpUtils.isPortAvailable(port)) {
+                    port = URLUtils.getRandomPort();
+                }
             }
             return buildLocalEnvironmentAddress(port);
         }

diff --git a/dinky-admin/src/main/java/org/dinky/service/impl/TaskServiceImpl.java b/dinky-admin/src/main/java/org/dinky/service/impl/TaskServiceImpl.java
@@ -225,7 +225,13 @@ public JobConfig buildJobConfig(TaskDTO task) {
                 config.setClusterId(jobInstance.getClusterId());
             }
         } else if (GatewayType.LOCAL.equalsValue(task.getType())) {
-            config.setClusterId(null);
+            if (task.getJobInstanceId() == null) {
+                config.setClusterId(null);
+            } else {
+                JobInstance jobInstance = jobInstanceService.getById(task.getJobInstanceId());
+                config.setClusterId(jobInstance.getClusterId());
+                config.setUseRemote(true);
+            }
             config.setClusterConfigurationId(null);
         } else {
             Optional.ofNullable(task.getClusterId()).ifPresent(config::setClusterId);

diff --git a/dinky-admin/src/main/java/org/dinky/utils/IpUtils.java b/dinky-admin/src/main/java/org/dinky/utils/IpUtils.java
@@ -22,6 +22,7 @@
 import org.dinky.assertion.Asserts;
 
 import java.net.InetAddress;
+import java.net.ServerSocket;
 import java.net.UnknownHostException;
 
 import javax.servlet.http.HttpServletRequest;
@@ -171,6 +172,16 @@ public static String getHostIp() {
         return "127.0.0.1";
     }
 
+    public static boolean isPortAvailable(int port) {
+        try (ServerSocket serverSocket = new ServerSocket(port)) {
+            // If the code reaches this point, the port is available
+            return true;
+        } catch (Exception e) {
+            // Port is not available
+            return false;
+        }
+    }
+
     public static String getHostName() {
         try {
             return InetAddress.getLocalHost().getHostName();

diff --git a/dinky-common/src/main/java/org/dinky/utils/URLUtils.java b/dinky-common/src/main/java/org/dinky/utils/URLUtils.java
@@ -107,6 +107,6 @@ public static String formatAddress(String webURL) {
 
     public static int getRandomPort() {
         Random random = new Random();
-        return random.nextInt(65536);
+        return 30000 + random.nextInt(35536);
     }
 }
diff --git a/dinky-core/src/main/java/org/dinky/job/JobConfig.java b/dinky-core/src/main/java/org/dinky/job/JobConfig.java
@@ -237,7 +237,7 @@ public void addGatewayConfig(Map<String, Object> config) {
     }
 
     public boolean isUseRemote() {
-        return !GatewayType.LOCAL.equalsValue(type);
+        return useRemote || !GatewayType.LOCAL.equalsValue(type);
     }
 
     public void buildLocal() {