This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 ccd85d2f5a [Fix][Zeta][Connector-V2] Cancel in-flight JDBC source
queries (#12419) (#12420)
ccd85d2f5a is described below
commit ccd85d2f5a048fcc4ae3b7f34b147acf5bdb2c42
Author: Jast <[email protected]>
AuthorDate: Mon Oct 12 06:07:32 2026 +0000
[Fix][Zeta][Connector-V2] Cancel in-flight JDBC source queries (#12419)
(#12420)
Co-authored-by: jast <[email protected]>
---
.../architecture/api-design/source-architecture.md | 11 ++
.../architecture/api-design/source-architecture.md | 3 +
.../apache/seatunnel/api/source/SourceReader.java | 24 ++++
.../seatunnel/jdbc/internal/JdbcInputFormat.java | 95 +++++++++++++-
.../seatunnel/jdbc/source/JdbcSourceReader.java | 5 +
.../jdbc/internal/JdbcInputFormatTest.java | 140 +++++++++++++++++++++
.../connectors/seatunnel/jdbc/JdbcMysqlIT.java | 84 +++++++++++++
.../test/resources/jdbc_mysql_source_cancel.conf | 42 +++++++
.../engine/server/TaskExecutionService.java | 28 +++++
.../seatunnel/engine/server/execution/Task.java | 3 +
.../engine/server/task/SeaTunnelTask.java | 31 ++++-
.../engine/server/task/flow/FlowLifeCycle.java | 2 +
.../server/task/flow/SourceFlowLifeCycle.java | 11 +-
13 files changed, 475 insertions(+), 4 deletions(-)
diff --git a/docs/en/architecture/api-design/source-architecture.md
b/docs/en/architecture/api-design/source-architecture.md
index 7ae73b1ec5..7ce518770a 100644
--- a/docs/en/architecture/api-design/source-architecture.md
+++ b/docs/en/architecture/api-design/source-architecture.md
@@ -367,6 +367,13 @@ public interface SourceReader<T, SplitT extends
SourceSplit>
*/
void open() throws Exception;
+ /**
+ * Request cancellation of an in-flight source operation (Zeta only).
+ * Called from an engine thread while open/pollNext/close may run
concurrently;
+ * implementations must be idempotent, thread-safe and lock-free.
+ */
+ default void cancel() {}
+
/**
* Poll next batch of records (non-blocking or timeout)
*/
@@ -415,6 +422,10 @@ public interface SourceReader<T, SplitT extends
SourceSplit>
- **Progress Tracking**: Track offset/position within each split
- **State Management**: Snapshot split progress for recovery
- **Split Management**: Handle split assignment, completion, and removal
+- **Job Cancellation**: Override `cancel()` when reads can block in an
external system (e.g. a
+ JDBC query). The engine calls it from a separate thread on job cancellation,
so implementations
+ must be idempotent, thread-safe, and must not acquire locks held by the
reader methods. Currently
+ only the Zeta engine invokes this hook.
**Implementation Example**:
diff --git a/docs/zh/architecture/api-design/source-architecture.md
b/docs/zh/architecture/api-design/source-architecture.md
index c8c8ff80e2..8fb4a7eb22 100644
--- a/docs/zh/architecture/api-design/source-architecture.md
+++ b/docs/zh/architecture/api-design/source-architecture.md
@@ -209,6 +209,9 @@ on addSplitsBack(splits):
- `addSplits(splits)`：接收枚举器分配的 splits
- `snapshotState(checkpointId)`：返回 split checkpoint state（实际接口返回
`List<SplitT>`）
- `handleNoMoreSplits()`：收到无更多 split 的信号
+- `cancel()`：请求中止阻塞在外部系统中的读取操作（目前仅 Zeta 引擎调用）。该方法由引擎线程在
+ `open`/`pollNext`/`close` 可能并发执行时调用，实现必须幂等、线程安全且不获取读取器持有的锁，
+ 并将取消请求视为粘性（阻塞操作开始前到达的取消仍需生效）
- `CheckpointListener` 回调：由框架触发 checkpoint 完成/中止通知
**关键职责**：
diff --git
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceReader.java
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceReader.java
index d095e426ef..f70662e3bb 100644
---
a/seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceReader.java
+++
b/seatunnel-api/src/main/java/org/apache/seatunnel/api/source/SourceReader.java
@@ -36,6 +36,30 @@ public interface SourceReader<T, SplitT extends SourceSplit>
/** Open the source reader. */
void open() throws Exception;
+ /**
+ * Request cancellation of an in-flight source operation, such as a query
blocked in an external
+ * system.
+ *
+ * <p>Contract for implementations:
+ *
+ * <ul>
+ * <li>Called from an engine thread that is not the task thread, while
{@link #open()}, {@link
+ * #pollNext(Collector)} or {@link #close()} may be running
concurrently.
+ * <li>May be invoked multiple times, before {@link #open()}, between
splits, or after {@link
+ * #close()}; implementations must be idempotent and thread-safe.
+ * <li>Must not acquire locks that the reader methods hold (e.g. the
checkpoint lock), or the
+ * cancellation would deadlock behind the very operation it is meant
to abort.
+ * <li>Must treat the request as sticky: a cancellation that arrives
before the blocking
+ * operation starts must still take effect.
+ * </ul>
+ *
+ * <p>The default implementation keeps existing source connectors
source-compatible. Sources
+ * that can block in an external client should override this method to
abort the external
+ * operation. Currently only the Zeta engine calls this hook; the Flink
and Spark translation
+ * layers do not.
+ */
+ default void cancel() {}
+
/**
* Called to close the reader, in case it holds on to any resources, like
threads or network
* connections.
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java
index b837827e5d..992a5c29dc 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormat.java
@@ -62,8 +62,14 @@ public class JdbcInputFormat implements Serializable {
private transient String splitTableId;
private transient TableSchema splitTableSchema;
- private transient PreparedStatement statement;
+ private transient volatile PreparedStatement statement;
private transient ResultSet resultSet;
+ // Runtime connection state written by the task thread and read by the
engine cancel thread;
+ // cancel() may close it while the task thread is still using it.
+ private transient volatile Connection activeConnection;
+ // Sticky cancellation flag set by the engine cancel thread and honoured
by cancel() and open(),
+ // so a cancel that arrives before the statement is published is not
silently lost.
+ private transient volatile boolean cancelled;
private volatile boolean hasNext;
public JdbcInputFormat(JdbcSourceConfig config, Map<TablePath,
CatalogTable> tables) {
@@ -91,6 +97,81 @@ public class JdbcInputFormat implements Serializable {
public void openInputFormat() {}
+ /**
+ * Cancels the JDBC operation currently owned by this input format.
+ *
+ * <p>This method is intentionally independent from {@link #close()}: the
engine may call it
+ * while the source task is blocked inside {@code executeQuery()} or
{@code ResultSet.next()}.
+ * Closing the statement and connection is the fallback for drivers that
do not implement {@link
+ * PreparedStatement#cancel()} correctly.
+ *
+ * <p>The cancellation is sticky: it is recorded in {@code cancelled} and
re-checked in {@link
+ * #open(JdbcSourceSplit)}, so a cancel that arrives before the statement
is published — or
+ * between splits — is not silently lost. Safe to call multiple times from
the engine cancel
+ * thread; only touches volatile fields and never acquires the checkpoint
lock.
+ */
+ public void cancel() {
+ if (cancelled) {
+ return;
+ }
+ cancelled = true;
+ abortInFlight();
+ }
+
+ /**
+ * Best-effort abort of the in-flight statement and its connection.
Idempotent and safe to run
+ * concurrently from the task thread (via {@link #ensureNotCancelled()})
and the cancel thread
+ * (via {@link #cancel()}), because JDBC drivers must tolerate repeated
{@code cancel()} and
+ * {@code close()} calls.
+ */
+ private void abortInFlight() {
+ PreparedStatement currentStatement = statement;
+ Connection currentConnection = activeConnection;
+ if (currentStatement != null) {
+ if (currentConnection == null) {
+ try {
+ currentConnection = currentStatement.getConnection();
+ } catch (SQLException e) {
+ LOG.warn("Get JDBC connection for cancelled statement
failed", e);
+ }
+ }
+ try {
+ currentStatement.cancel();
+ } catch (SQLException e) {
+ LOG.warn("Cancel JDBC statement failed", e);
+ }
+ try {
+ currentStatement.close();
+ } catch (SQLException e) {
+ LOG.warn("Close cancelled JDBC statement failed", e);
+ }
+ }
+ if (currentConnection != null) {
+ try {
+ currentConnection.close();
+ } catch (SQLException e) {
+ LOG.warn("Close cancelled JDBC connection failed", e);
+ }
+ }
+ }
+
+ /**
+ * Honours a sticky cancellation before more work is started. Checked at
the start of {@link
+ * #open(JdbcSourceSplit)} (a cancelled reader must not connect for the
next split) and again
+ * after the statement is published (the usual publish-then-check
ordering, so a cancel racing
+ * with statement creation is always observed by at least one side).
+ */
+ private void ensureNotCancelled() throws IOException {
+ if (!cancelled) {
+ return;
+ }
+ abortInFlight();
+ statement = null;
+ activeConnection = null;
+ throw new IOException(
+ "JDBC source reader has been cancelled, stop opening split " +
splitTableId);
+ }
+
public void closeInputFormat() throws IOException {
try {
close();
@@ -109,11 +190,18 @@ public class JdbcInputFormat implements Serializable {
* @throws IOException if there's an error during the execution of the
query
*/
public void open(JdbcSourceSplit inputSplit) throws IOException {
+ ensureNotCancelled();
try {
splitTableSchema =
tables.get(inputSplit.getTablePath()).getTableSchema();
splitTableId = inputSplit.getTablePath().toString();
statement = chunkSplitter.generateSplitStatement(inputSplit,
splitTableSchema);
+ try {
+ activeConnection = statement.getConnection();
+ } catch (SQLException e) {
+ LOG.debug("JDBC statement did not expose its connection before
execution", e);
+ }
+ ensureNotCancelled();
resultSet = statement.executeQuery();
hasNext = resultSet.next();
} catch (SQLException se) {
@@ -152,6 +240,9 @@ public class JdbcInputFormat implements Serializable {
*/
public void close() throws IOException {
Connection connection = getStatementConnection();
+ if (connection == null) {
+ connection = activeConnection;
+ }
if (resultSet != null) {
try {
resultSet.close();
@@ -174,6 +265,7 @@ public class JdbcInputFormat implements Serializable {
hasNext = false;
splitTableSchema = null;
splitTableId = null;
+ activeConnection = null;
finishReadTransaction(connection);
}
@@ -183,6 +275,7 @@ public class JdbcInputFormat implements Serializable {
}
try {
Connection connection = statement.getConnection();
+ activeConnection = connection;
if (connection == null) {
LOG.warn(
"The JDBC source statement returned no connection. "
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceReader.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceReader.java
index 40bbe28d97..2ad316d446 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceReader.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/JdbcSourceReader.java
@@ -57,6 +57,11 @@ public class JdbcSourceReader implements
SourceReader<SeaTunnelRow, JdbcSourceSp
inputFormat.openInputFormat();
}
+ @Override
+ public void cancel() {
+ inputFormat.cancel();
+ }
+
@Override
public void close() throws IOException {
inputFormat.closeInputFormat();
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormatTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormatTest.java
index 13f5fb72f5..dd692ae9b0 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormatTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/JdbcInputFormatTest.java
@@ -36,8 +36,18 @@ import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Collections;
import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
@@ -212,6 +222,136 @@ class JdbcInputFormatTest {
verify(context.chunkSplitter).close();
}
+ @Test
+ void shouldCancelActiveStatementAndConnection() throws Exception {
+ TestContext context = createContext(true);
+ context.openEmptySplit();
+
+ context.inputFormat.cancel();
+ context.inputFormat.cancel();
+
+ verify(context.statement, times(1)).cancel();
+ verify(context.statement, times(1)).close();
+ verify(context.connection, times(1)).close();
+ }
+
+ @Test
+ void shouldCancelStatementWhileExecuteQueryIsInFlight() throws Exception {
+ TestContext context = createContext(true);
+ CountDownLatch executeStarted = new CountDownLatch(1);
+ CountDownLatch cancelObserved = new CountDownLatch(1);
+ when(context.chunkSplitter.generateSplitStatement(SPLIT, TABLE_SCHEMA))
+ .thenReturn(context.statement);
+ when(context.statement.getConnection()).thenReturn(context.connection);
+ when(context.resultSet.next()).thenReturn(false);
+ doAnswer(
+ invocation -> {
+ executeStarted.countDown();
+ assertTrue(cancelObserved.await(5,
TimeUnit.SECONDS));
+ return context.resultSet;
+ })
+ .when(context.statement)
+ .executeQuery();
+ doAnswer(
+ invocation -> {
+ cancelObserved.countDown();
+ return null;
+ })
+ .when(context.statement)
+ .cancel();
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> openFuture =
+ executor.submit(
+ (Callable<Void>)
+ () -> {
+ context.inputFormat.open(SPLIT);
+ return null;
+ });
+ assertTrue(executeStarted.await(5, TimeUnit.SECONDS));
+ context.inputFormat.cancel();
+ assertTrue(cancelObserved.await(5, TimeUnit.SECONDS));
+ openFuture.get(5, TimeUnit.SECONDS);
+ verify(context.statement).cancel();
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * Regression test for the lost-cancel window: a cancel recorded before
{@code open()} must be
+ * honoured, so the reader neither connects for the next split nor starts
a new query.
+ */
+ @Test
+ void shouldNotOpenNextSplitAfterCancel() throws Exception {
+ TestContext context = createContext(true);
+
+ context.inputFormat.cancel();
+
+ assertThrows(IOException.class, () -> context.inputFormat.open(SPLIT));
+ verify(context.chunkSplitter, never()).generateSplitStatement(SPLIT,
TABLE_SCHEMA);
+ verify(context.statement, never()).executeQuery();
+ }
+
+ /**
+ * Regression test for the publish-then-check ordering: a cancel that
lands after the statement
+ * is published but before the query starts must abort the statement
instead of being lost.
+ */
+ @Test
+ void shouldAbortOpenWhenCancelRacesWithStatementCreation() throws
Exception {
+ TestContext context = createContext(true);
+ AtomicBoolean cancelInjected = new AtomicBoolean();
+ when(context.chunkSplitter.generateSplitStatement(SPLIT, TABLE_SCHEMA))
+ .thenReturn(context.statement);
+ doAnswer(
+ invocation -> {
+ // Simulates the engine cancel thread winning the
race after the
+ // statement is published but before
executeQuery() starts. The guard
+ // keeps the nested abortInFlight() call from
re-entering the cancel.
+ if (cancelInjected.compareAndSet(false, true)) {
+ context.inputFormat.cancel();
+ }
+ return context.connection;
+ })
+ .when(context.statement)
+ .getConnection();
+ when(context.resultSet.next()).thenReturn(false);
+
+ assertThrows(IOException.class, () -> context.inputFormat.open(SPLIT));
+
+ verify(context.statement, never()).executeQuery();
+ verify(context.statement, atLeastOnce()).cancel();
+ verify(context.statement, atLeastOnce()).close();
+ verify(context.connection, atLeastOnce()).close();
+ }
+
+ @Test
+ void shouldStillCloseConnectionWhenStatementCancelFails() throws Exception
{
+ TestContext context = createContext(true);
+ context.openEmptySplit();
+ doThrow(new SQLException("cancel
rejected")).when(context.statement).cancel();
+
+ context.inputFormat.cancel();
+
+ verify(context.statement).close();
+ verify(context.connection).close();
+ }
+
+ @Test
+ void shouldAllowCloseAfterCancel() throws Exception {
+ TestContext context = createContext(true);
+ context.openEmptySplit();
+ when(context.connection.isClosed()).thenReturn(false);
+ when(context.connection.getAutoCommit()).thenReturn(false);
+
+ context.inputFormat.cancel();
+ context.inputFormat.close();
+
+ verify(context.connection, times(1)).rollback();
+ verify(context.connection, times(1)).setAutoCommit(true);
+ }
+
@Test
void shouldIgnoreClosedConnection() throws SQLException {
Connection connection = mock(Connection.class);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlIT.java
index 60272e2135..1a5263067a 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlIT.java
@@ -52,11 +52,14 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceFactory;
import org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceSplit;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.source.JdbcSourceSplitEnumerator;
import org.apache.seatunnel.connectors.seatunnel.jdbc.state.JdbcSourceState;
+import org.apache.seatunnel.e2e.common.container.EngineType;
import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
import org.jetbrains.annotations.NotNull;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
import org.testcontainers.containers.Container;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.MySQLContainer;
@@ -86,6 +89,12 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.awaitility.Awaitility.given;
public class JdbcMysqlIT extends AbstractJdbcIT {
@@ -113,6 +122,8 @@ public class JdbcMysqlIT extends AbstractJdbcIT {
"/jdbc_mysql_source_and_sink_parallel_upper_lower.conf",
"/jdbc_mysql_source_and_sink.sql",
"/jdbc_mysql_source_and_sink_parallel.sql");
+ private static final String CANCEL_CONFIG_FILE =
"/jdbc_mysql_source_cancel.conf";
+ private static final String CANCEL_QUERY_MARKER =
"seatunnel-jdbc-cancel-e2e";
private static final String CREATE_SQL =
"CREATE TABLE IF NOT EXISTS %s\n"
+ "(\n"
@@ -472,6 +483,79 @@ public class JdbcMysqlIT extends AbstractJdbcIT {
testTinyInt1AsBooleanOrTINYINT(false, BasicType.BYTE_TYPE);
}
+ @TestTemplate
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason =
+ "TestContainer.executeJob/cancelJob/getJobStatus are only
implemented by the "
+ + "Zeta engine, and the tested cancel hook is Zeta
only")
+ public void testCancelJdbcSourceQuery(TestContainer container) throws
Exception {
+ // The Zeta client requires --set-job-id to be numeric, and the job
status/cancel REST
+ // calls address the job by the same id.
+ String jobId = String.valueOf(System.nanoTime());
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ Future<Container.ExecResult> jobFuture =
+ executor.submit(() -> container.executeJob(CANCEL_CONFIG_FILE,
jobId));
+
+ try {
+ given().ignoreExceptions()
+ .await()
+ .atMost(2, TimeUnit.MINUTES)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ "RUNNING",
container.getJobStatus(jobId)));
+ given().await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() ->
Assertions.assertTrue(hasRunningCancelQuery()));
+
+ Container.ExecResult cancelResult = container.cancelJob(jobId);
+ Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
+
+ given().await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() ->
Assertions.assertFalse(hasRunningCancelQuery()));
+
+ given().ignoreExceptions()
+ .await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ "CANCELED",
container.getJobStatus(jobId)));
+ jobFuture.get(30, TimeUnit.SECONDS);
+ } finally {
+ if (!jobFuture.isDone()) {
+ container.cancelJob(jobId);
+ }
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * Checks the MySQL process list for the in-flight source query,
identified by the marker
+ * comment in {@code jdbc_mysql_source_cancel.conf}. Excludes this control
connection itself so
+ * the helper cannot match its own {@code PROCESSLIST} statement.
Connection errors are
+ * propagated so they fail with their message instead of a blind timeout.
+ */
+ private boolean hasRunningCancelQuery() throws SQLException {
+ String processListQuery =
+ "SELECT INFO FROM information_schema.PROCESSLIST "
+ + "WHERE ID <> CONNECTION_ID() AND INFO LIKE '%"
+ + CANCEL_QUERY_MARKER
+ + "%'";
+ try (Connection controlConnection =
+ DriverManager.getConnection(
+ jdbcCase.getJdbcUrl().replace(HOST,
dbServer.getHost()),
+ jdbcCase.getUserName(),
+ jdbcCase.getPassword());
+ Statement statement = controlConnection.createStatement();
+ ResultSet resultSet =
statement.executeQuery(processListQuery)) {
+ return resultSet.next();
+ }
+ }
+
private void testTinyInt1AsBooleanOrTINYINT(boolean intTypeNarrowing,
BasicType<?> exceptType)
throws SQLException {
try (MySqlCatalog catalogWithIntTypeNarrowing =
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_source_cancel.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_source_cancel.conf
new file mode 100644
index 0000000000..e7dff29197
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_source_cancel.conf
@@ -0,0 +1,42 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ jdbc {
+ url = "jdbc:mysql://mysql-e2e:3306/seatunnel"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "root"
+ password = "Abc!@#135_seatunnel"
+ table_path = "seatunnel.source"
+ # Single-split read: skips the split metadata queries (min/max) that would
otherwise also run
+ # the sleep() predicate and stall split enumeration for minutes.
+ enable_concurrent_read = false
+ where_condition = "where sleep(60) = 0 /* seatunnel-jdbc-cancel-e2e */"
+ properties {
+ useSSL = false
+ }
+ }
+}
+
+sink {
+ Console {}
+}
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 d72723cddc..94b52b55e8 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
@@ -94,6 +94,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -1572,6 +1573,33 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
* Cancels tasks and background work owned by this tracker, regardless
of active generation.
*/
private void cancelAllTask() {
+ try {
+ ExecutorService tracingExecutorService =
MDCTracer.tracing(executorService);
+ taskGroup
+ .getTasks()
+ .forEach(
+ task ->
+ tracingExecutorService.submit(
+ () -> {
+ try {
+ task.cancel();
+ } catch (Throwable t) {
+ logger.warning(
+ String.format(
+
"Cancel task %s external resources failed: %s",
+
task.getTaskID(),
+
ExceptionUtils.getMessage(
+
t)));
+ }
+ }));
+ } catch (RejectedExecutionException e) {
+ // The executor may already be shut down; never let this block
the interrupt-based
+ // cancellation below.
+ logger.warning(
+ "Submitting task external resource cancellation
failed, the executor may "
+ + "have been shut down: "
+ + ExceptionUtils.getMessage(e));
+ }
try {
blockingFutures.forEach(f -> f.cancel(true));
currRunningTaskFuture.values().forEach(f -> f.cancel(true));
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/Task.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/Task.java
index 5a771c66e2..5d413b6b9a 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/Task.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/execution/Task.java
@@ -45,6 +45,9 @@ public interface Task
return false;
}
+ /** Request cancellation of external blocking resources owned by this
task. */
+ default void cancel() {}
+
default void close() throws IOException {}
default void setTaskExecutionContext(TaskExecutionContext
taskExecutionContext) {}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelTask.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelTask.java
index 6c136d8e66..8fdef7d0e2 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelTask.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/SeaTunnelTask.java
@@ -79,6 +79,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
@@ -121,7 +122,13 @@ public abstract class SeaTunnelTask extends AbstractTask {
protected FlowLifeCycle startFlowLifeCycle;
- protected List<FlowLifeCycle> allCycles;
+ // Volatile and copy-on-write: cancel() iterates this list from the engine
cancel thread while
+ // init() may still be filling it on the task thread.
+ protected volatile List<FlowLifeCycle> allCycles;
+
+ // Set by cancel() when it ran before the life cycles were built; init()
replays the cancel in
+ // that case so no flow keeps blocking in an external system.
+ private volatile boolean cancelRequested;
protected List<OneInputFlowLifeCycle<Record<?>>> outputs;
@@ -182,11 +189,16 @@ public abstract class SeaTunnelTask extends AbstractTask {
observabilityEnabled = resolveObservabilityEnabled();
this.currState = SeaTunnelTaskState.INIT;
flowFutures = new ArrayList<>();
- allCycles = new ArrayList<>();
+ allCycles = new CopyOnWriteArrayList<>();
startFlowLifeCycle = convertFlowToActionLifeCycle(executionFlow);
for (FlowLifeCycle cycle : allCycles) {
cycle.init();
}
+ // A cancel that raced with this init() may have missed cycles that
were still being built;
+ // replay it so no reader keeps blocking in an external system after
cancellation.
+ if (cancelRequested) {
+ cancel();
+ }
CompletableFuture.allOf(flowFutures.toArray(new CompletableFuture[0]))
.whenComplete((s, e) -> closeCalled = true);
}
@@ -403,6 +415,21 @@ public abstract class SeaTunnelTask extends AbstractTask {
protected abstract void collect() throws Exception;
+ /**
+ * Cancels external blocking resources owned by every flow life cycle of
this task.
+ *
+ * <p>Called from the engine cancel thread, potentially while {@link
#init()} is still building
+ * {@code allCycles} on the task thread, so the request is recorded in
{@code cancelRequested}
+ * and replayed at the end of {@link #init()}. Must stay cheap, idempotent
and lock-free.
+ */
+ @Override
+ public void cancel() {
+ cancelRequested = true;
+ if (allCycles != null) {
+ allCycles.forEach(FlowLifeCycle::cancel);
+ }
+ }
+
@Override
public Set<URL> getJarsUrl() {
return getFlowInfo((action, set) -> set.addAll(action.getJarUrls()));
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/FlowLifeCycle.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/FlowLifeCycle.java
index 59ba4db3b1..230c232ff7 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/FlowLifeCycle.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/FlowLifeCycle.java
@@ -25,6 +25,8 @@ public interface FlowLifeCycle {
default void open() throws Exception {}
+ default void cancel() {}
+
default void close() throws IOException {}
default void hook() throws IOException {}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java
index f8fdc17567..54c1e9af68 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SourceFlowLifeCycle.java
@@ -102,7 +102,9 @@ public class SourceFlowLifeCycle<T, SplitT extends
SourceSplit> extends ActionFl
private Address enumeratorTaskAddress;
- private SourceReader<T, SplitT> reader;
+ // Volatile: cancel() reads it from the engine cancel thread while the
task thread may still be
+ // creating the reader in init().
+ private volatile SourceReader<T, SplitT> reader;
private transient Serializer<SplitT> splitSerializer;
@@ -185,6 +187,13 @@ public class SourceFlowLifeCycle<T, SplitT extends
SourceSplit> extends ActionFl
register();
}
+ @Override
+ public void cancel() {
+ if (reader != null) {
+ reader.cancel();
+ }
+ }
+
/**
* Timer callback invoked by the {@code timerFlushWorker} thread pool.
*
