[Improve][Zeta] Remove misleading exception log when job be canceled (#8988)

Author: Hisoka-X

Diff for: seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/common/container/seatunnel/SeaTunnelContainer.java

@@ -159,8 +159,8 @@ protected GenericContainer<?> createSeaTunnelContainerWithFakeSourceAndInMemoryS
                                                 "seatunnel-engine:" + JDK_DOCKER_IMAGE)))
                         .waitingFor(Wait.forLogMessage(".*received new worker register:.*", 1));
         copySeaTunnelStarterToContainer(server);
-        server.setPortBindings(Collections.singletonList("5801:5801"));
-        server.setExposedPorts(Collections.singletonList(5801));
+        server.setPortBindings(Arrays.asList("5801:5801", "8080:8080"));
+        server.setExposedPorts(Arrays.asList(5801, 8080));
 
         server.withCopyFileToContainer(
                 MountableFile.forHostPath(

Diff for: seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/sink/inmemory/InMemorySinkFactory.java

@@ -40,6 +40,9 @@ public class InMemorySinkFactory
     public static final Option<Boolean> THROW_EXCEPTION =
             Options.key("throw_exception").booleanType().defaultValue(false);
 
+    public static final Option<Boolean> WRITER_SLEEP =
+            Options.key("writer_sleep").booleanType().defaultValue(false);
+
     public static final Option<Boolean> THROW_OUT_OF_MEMORY =
             Options.key("throw_out_of_memory").booleanType().defaultValue(false);
     public static final Option<Boolean> CHECKPOINT_SLEEP =
@@ -66,6 +69,7 @@ public OptionRule optionRule() {
                 .optional(
                         THROW_EXCEPTION,
                         THROW_OUT_OF_MEMORY,
+                        WRITER_SLEEP,
                         CHECKPOINT_SLEEP,
                         THROW_EXCEPTION_OF_COMMITTER,
                         ASSERT_OPTIONS_KEY,

Diff for: seatunnel-e2e/seatunnel-e2e-common/src/test/java/org/apache/seatunnel/e2e/sink/inmemory/InMemorySinkWriter.java

@@ -79,6 +79,14 @@ public InMemorySinkWriter(ReadonlyConfig config) {
 
     @Override
     public void write(SeaTunnelRow element) throws IOException {
+        if (config.get(InMemorySinkFactory.WRITER_SLEEP)) {
+            try {
+                Thread.sleep(999999999L);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
         if (config.get(InMemorySinkFactory.THROW_OUT_OF_MEMORY)) {
             throw new OutOfMemoryError();
         }

Diff for: seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/JobClientJobProxyIT.java

@@ -24,10 +24,16 @@
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.Container;
 
+import lombok.extern.slf4j.Slf4j;
+
 import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 
 import static org.apache.seatunnel.e2e.common.util.ContainerUtil.PROJECT_ROOT_PATH;
+import static org.testcontainers.shaded.org.awaitility.Awaitility.given;
 
+@Slf4j
 public class JobClientJobProxyIT extends SeaTunnelEngineContainer {
 
     @Override
@@ -73,6 +79,44 @@ public void testNoDuplicatedReleaseSlot() throws IOException, InterruptedExcepti
                 server.getLogs().contains("wrong target release operation with job"));
     }
 
+    @Test
+    public void testNoExceptionLogWhenCancelJob() throws IOException, InterruptedException {
+        String jobId = String.valueOf(System.currentTimeMillis());
+        CompletableFuture.runAsync(
+                () -> {
+                    try {
+                        executeJob(
+                                "/stream_fakesource_to_inmemory_pending_row_in_queue.conf", jobId);
+                    } catch (Exception e) {
+                        log.error("Commit task exception :" + e.getMessage());
+                        throw new RuntimeException();
+                    }
+                });
+
+        given().pollDelay(10, TimeUnit.SECONDS)
+                .await()
+                .pollDelay(5000L, TimeUnit.MILLISECONDS)
+                .untilAsserted(
+                        () -> {
+                            Assertions.assertEquals("RUNNING", this.getJobStatus(jobId));
+                        });
+
+        String logBeforeCancel = this.getServerLogs();
+        cancelJob(jobId);
+        given().pollDelay(10, TimeUnit.SECONDS)
+                .await()
+                .pollDelay(5000L, TimeUnit.MILLISECONDS)
+                .untilAsserted(
+                        () -> {
+                            Assertions.assertEquals("CANCELED", this.getJobStatus(jobId));
+                        });
+        String logAfterCancel = this.getServerLogs().substring(logBeforeCancel.length());
+        // in TaskExecutionService.BlockingWorker::run catch Throwable
+        Assertions.assertFalse(logAfterCancel.contains("Exception in"), logAfterCancel);
+        Assertions.assertEquals(
+                4, StringUtils.countMatches(logAfterCancel, "Interrupted task"), logAfterCancel);
+    }
+
     @Test
     public void testMultiTableSinkFailedWithThrowable() throws IOException, InterruptedException {
         Container.ExecResult execResult =

Diff for: seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/seatunnel_fixed_slot_num.yaml

@@ -33,3 +33,6 @@ seatunnel:
         max-retained: 3
         plugin-config:
           namespace: /tmp/seatunnel/checkpoint_snapshot/
+    http:
+      enable-http: true
+      port: 8080

Diff for: seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/stream_fakesource_to_inmemory_pending_row_in_queue.conf

@@ -0,0 +1,47 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+  job.mode = "STREAMING"
+  checkpoint.interval = 5000
+}
+
+source {
+  # This is a example source plugin **only for test and demonstrate the feature source plugin**
+  FakeSource {
+    # More than TaskGroupWithIntermediateBlockingQueue::QUEUE_SIZE
+    row.num = 9999
+    parallelism = 1
+    schema = {
+      fields {
+        c_int = int
+      }
+    }
+  }
+}
+
+transform {
+}
+
+sink {
+  InMemory {
+    writer_sleep = true
+  }
+}

Diff for: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java

@@ -702,7 +702,11 @@ public void run() {
                     taskGroupExecutionTracker.exception(e);
                 }
             } catch (Throwable e) {
-                logger.warning("Exception in " + t, e);
+                if (taskGroupExecutionTracker.isCancel.get()) {
+                    logger.warning(String.format("Interrupted task %d - %s", t.getTaskID(), t));
+                } else {
+                    logger.warning("Exception in " + t, e);
+                }
                 taskGroupExecutionTracker.exception(e);
             } finally {
                 taskGroupExecutionTracker.taskDone(t);