[INLONG-4212][Manager] The processor executor maybe throws a null poi…

…nter exception (#4213)
apache · May 16, 2022 · b8bec82 · b8bec82
1 parent c1e3bc3
commit b8bec82
Showing 1 changed file with 18 additions and 14 deletions.
diff --git a/...low/src/main/java/org/apache/inlong/manager/workflow/core/impl/ProcessorExecutorImpl.java b/...low/src/main/java/org/apache/inlong/manager/workflow/core/impl/ProcessorExecutorImpl.java
@@ -103,8 +103,10 @@ public void executeStart(Element element, WorkflowContext context) {
 
         // If it is a continuous task execution transaction isolation
         if (element instanceof WorkflowTask) {
-            transactionHelper.execute(executeCompleteInTransaction(element, context),
-                    TransactionDefinition.PROPAGATION_NESTED);
+            TransactionCallback<Object> callback = executeCompleteInTransaction(element, context);
+            if (callback != null) {
+                transactionHelper.execute(callback, TransactionDefinition.PROPAGATION_NESTED);
+            }
             return;
         }
 
@@ -120,7 +122,9 @@ public void executeComplete(Element element, WorkflowContext context) {
             return;
         }
         List<Element> nextElements = processor.next(element, context);
-        nextElements.forEach(next -> executeStart(next, context));
+        for (Element next : nextElements) {
+            executeStart(next, context);
+        }
     }
 
     private boolean isSkipCurrentElement(Element element, WorkflowContext context) {
@@ -150,20 +154,20 @@ private void executeSkipAndNext(Element element, WorkflowContext context) {
         // Execute next
         context.getActionContext().setAction(((NextableElement) element).defaultNextAction());
         List<Element> nextElements = processor.next(element, context);
-        nextElements.forEach(next -> executeStart(next, context));
+        for (Element next : nextElements) {
+            executeStart(next, context);
+        }
     }
 
     private TransactionCallback<Object> executeCompleteInTransaction(Element element, WorkflowContext context) {
-        return s -> {
-            try {
-                executeComplete(element, context);
-                return null;
-            } catch (WorkflowNoRollbackException e) { // Exception does not roll back
-                throw e;
-            } catch (Exception e) { // The exception is only rolled back once
-                throw new WorkflowRollbackOnceException(e.getMessage());
-            }
-        };
+        try {
+            executeComplete(element, context);
+            return null;
+        } catch (WorkflowNoRollbackException e) { // Exception does not roll back
+            throw e;
+        } catch (Exception e) { // The exception is only rolled back once
+            throw new WorkflowRollbackOnceException(e.getMessage());
+        }
     }
 
 }