This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12420-a4c291d8fd6c7b785b90334d4ec2ee68e026dd7a in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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. *
