[Fix-4263] Fix web socket bug in data studio (#4268)

Author: aiwenmo

dinky-admin/src/main/java/org/dinky/aop/UdfClassLoaderAspect.java

@@ -19,6 +19,7 @@
 
 package org.dinky.ws.handler;
 
+import org.dinky.assertion.Asserts;
 import org.dinky.ws.GlobalWebSocketTopic;
 import org.dinky.ws.WsSendEvent;
 
@@ -50,11 +51,13 @@ public void run() {
                     @Override
                     public void run() {
                         Map<String, ?> data = autoMessageSend();
-                        WsSendEvent event = WsSendEvent.builder()
-                                .topic(topic)
-                                .paramsAndData(data)
-                                .build();
-                        applicationEventPublisher.publishEvent(event);
+                        if (Asserts.isNotNullMap(data)) {
+                            WsSendEvent event = WsSendEvent.builder()
+                                    .topic(topic)
+                                    .paramsAndData(data)
+                                    .build();
+                            applicationEventPublisher.publishEvent(event);
+                        }
                     }
                 },
                 0,

dinky-admin/src/main/java/org/dinky/ws/handler/ScheduleMessageEventHandler.java

@@ -25,6 +25,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentSkipListSet;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.stereotype.Service;
@@ -34,7 +35,8 @@
 @Slf4j
 @Service
 public class TaskRunInstance extends ScheduleMessageEventHandler {
-    // todo 这里需要优化下，应该触发hook调用
+
+    private Set<Integer> runningJobIds = new ConcurrentSkipListSet<>();
 
     @Override
     public Map<String, Object> firstSubscribe(Set<String> allParams) {
@@ -51,11 +53,14 @@ public GlobalWebSocketTopic getTopic() {
     @Override
     public Map<String, Object> autoMessageSend() {
         Set<Integer> currentMonitorTaskIds = FlinkJobThreadPool.getInstance().getCurrentMonitorTaskIds();
-        Map<String, Object> result = new HashMap<>();
-        if (!currentMonitorTaskIds.isEmpty()) {
+        if (!runningJobIds.equals(currentMonitorTaskIds)) {
+            runningJobIds.clear();
+            runningJobIds.addAll(currentMonitorTaskIds);
+            Map<String, Object> result = new HashMap<>();
             result.put("RunningTaskId", currentMonitorTaskIds);
+            return result;
         }
-        return result;
+        return new HashMap<>();
     }
 
     @Override