[seatunnel] branch dev updated: [Fix][Zeta] Fix Zeta will close task twice error (#5422)

[gaojun2048]

This is an automated email from the ASF dual-hosted git repository.

gaojun2048 pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new d1e9673f76 [Fix][Zeta] Fix Zeta will close task twice error (#5422)
d1e9673f76 is described below

commit d1e9673f76d3692beee3bef0adef338dd521d822
Author: Jia Fan <fanjiaemi...@qq.com>
AuthorDate: Wed Sep 6 14:42:42 2023 +0800

    [Fix][Zeta] Fix Zeta will close task twice error (#5422)
---
 .../apache/seatunnel/engine/server/TaskExecutionService.java | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git 
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
 
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
index 2725836f87..009225347c 100644
--- 
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
+++ 
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
@@ -603,10 +603,10 @@ public class TaskExecutionService implements 
DynamicMetricsProvider {
             ClassLoader oldClassLoader = 
Thread.currentThread().getContextClassLoader();
             Thread.currentThread().setContextClassLoader(classLoader);
             final Task t = tracker.task;
+            ProgressState result = null;
             try {
                 startedLatch.countDown();
                 t.init();
-                ProgressState result;
                 do {
                     result = t.call();
                 } while (!result.isDone()
@@ -623,10 +623,12 @@ public class TaskExecutionService implements 
DynamicMetricsProvider {
                 taskGroupExecutionTracker.exception(e);
             } finally {
                 taskGroupExecutionTracker.taskDone(t);
-                try {
-                    tracker.task.close();
-                } catch (IOException e) {
-                    logger.severe("Close task error", e);
+                if (result == null || !result.isDone()) {
+                    try {
+                        tracker.task.close();
+                    } catch (IOException e) {
+                        logger.severe("Close task error", e);
+                    }
                 }
             }
             Thread.currentThread().setContextClassLoader(oldClassLoader);