From 6bf286a52235689a01eab48af71b144df68ea38a Mon Sep 17 00:00:00 2001
From: lenboo <baoliang.leon@gmail.com>
Date: Sun, 5 Sep 2021 11:54:37 +0800
Subject: [PATCH] fix critical bugs

---
 .../queue/StateEventResponseService.java      | 22 +++++++++++--------
 1 file changed, 13 insertions(+), 9 deletions(-)

diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/StateEventResponseService.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/StateEventResponseService.java
index 1ee053c0d..2c84afa50 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/StateEventResponseService.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/StateEventResponseService.java
@@ -111,8 +111,8 @@ public class StateEventResponseService {
                     // if not task , blocking here
                     StateEvent stateEvent = eventQueue.take();
                     persist(stateEvent);
-                } catch (Exception e) {
-                    logger.error("persist task error", e);
+                } catch (InterruptedException e) {
+                    logger.warn("persist task error", e);
                 }
             }
             logger.info("StateEventResponseWorker stopped");
@@ -128,14 +128,18 @@ public class StateEventResponseService {
     }
 
     private void persist(StateEvent stateEvent) {
-        if (!this.processInstanceMapper.containsKey(stateEvent.getProcessInstanceId())) {
-            writeResponse(stateEvent, ExecutionStatus.FAILURE);
-            return;
-        }
+        try {
+            if (!this.processInstanceMapper.containsKey(stateEvent.getProcessInstanceId())) {
+                writeResponse(stateEvent, ExecutionStatus.FAILURE);
+                return;
+            }
 
-        WorkflowExecuteThread workflowExecuteThread = this.processInstanceMapper.get(stateEvent.getProcessInstanceId());
-        workflowExecuteThread.addStateEvent(stateEvent);
-        writeResponse(stateEvent, ExecutionStatus.SUCCESS);
+            WorkflowExecuteThread workflowExecuteThread = this.processInstanceMapper.get(stateEvent.getProcessInstanceId());
+            workflowExecuteThread.addStateEvent(stateEvent);
+            writeResponse(stateEvent, ExecutionStatus.SUCCESS);
+        } catch (Exception e) {
+            logger.error("persist event queue error:", stateEvent.toString(), e);
+        }
     }
 
     public BlockingQueue<StateEvent> getEventQueue() {
-- 
GitLab