From e2c9073f2f50a6f6c5e7b5e1172276b17cabe54b Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Mon, 18 May 2026 18:09:07 +0800 Subject: [PATCH 01/23] =?UTF-8?q?fix(dataset):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E5=AE=9A=E6=97=B6=E5=90=8C=E6=AD=A5=E4=BB=BB=E5=8A=A1=20JobDat?= =?UTF-8?q?aMap=20key=20=E4=B8=8D=E4=B8=80=E8=87=B4=E5=AF=BC=E8=87=B4=20NP?= =?UTF-8?q?E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../edurt/datacap/service/initializer/job/DatasetJob.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/initializer/job/DatasetJob.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/initializer/job/DatasetJob.java index c63e2e01ff..883f83aaba 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/initializer/job/DatasetJob.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/initializer/job/DatasetJob.java @@ -25,7 +25,12 @@ public void setService(DataSetService service) @Override protected void executeInternal(JobExecutionContext context) { - String code = context.getJobDetail().getJobDataMap().get("code").toString(); + Object idValue = context.getJobDetail().getJobDataMap().get("id"); + if (idValue == null) { + log.warn("Job [ {} ] skipped: missing 'id' in JobDataMap", context.getJobDetail().getKey()); + return; + } + String code = idValue.toString(); log.info("Job [ {} ] run time [ {} ]", code, context.getFireTime().getTime()); this.service.syncData(code); } From e4f3a489e1af075b9a45e80c872cc5f428c91cc9 Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Wed, 20 May 2026 12:52:11 +0800 Subject: [PATCH 02/23] =?UTF-8?q?feat(executor-local):=20=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E6=B5=81=E5=BC=8F=E5=90=8C=E6=AD=A5=E5=B9=B6=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=20OOM=20/=20SQL=20=E6=B3=A8=E5=85=A5=20/=20NULL=20?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 SPI: RowCallback / BatchWriter / JdbcStreamAdapter，JDBC 插件默认获得流式读 + PreparedStatement 批量写能力 - PluginService 增加 supportsStreaming / executeStream / openBatchWriter 默认方法，旧插件零改动 - LocalExecutorService 两端 JDBC 时走流式路径，源端 fetchSize 拉取、目标端 batchSize 批量 flush，全程不物化结果集 - 回退路径修复 NULL/类型/转义错误（替换已废弃的 commons-lang2 escapeSql），按 batch 切片提交不再拼巨大字符串 - 新增 datacap.executor.local.fetchSize / batchSize 配置项，默认 1000 --- configure/docker/application.properties | 6 + configure/etc/conf/application.properties | 6 + .../service/impl/DataSetServiceImpl.java | 6 +- .../io/edurt/datacap/spi/PluginService.java | 48 +++ .../datacap/spi/adapter/BatchWriter.java | 14 + .../spi/adapter/JdbcStreamAdapter.java | 306 +++++++++++++++++ .../datacap/spi/adapter/RowCallback.java | 18 + .../executor/local/LocalExecutorService.kt | 324 ++++++++++++++---- .../executor/configure/ExecutorRequest.kt | 6 +- 9 files changed, 671 insertions(+), 63 deletions(-) create mode 100644 core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/BatchWriter.java create mode 100644 core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcStreamAdapter.java create mode 100644 core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/RowCallback.java diff --git a/configure/docker/application.properties b/configure/docker/application.properties index 89441658e1..8985e7c9bc 100644 --- a/configure/docker/application.properties +++ b/configure/docker/application.properties @@ -48,6 +48,12 @@ datacap.executor.engine=SPARK datacap.executor.startScript=start-seatunnel-spark-connector-v2.sh datacap.executor.seatunnel.home=/opt/lib/seatunnel +# ------ Local Executor (in-process JDBC pipeline) ------ # +# JDBC fetchSize for streaming reads from the source. MySQL / MariaDB override to MIN_VALUE automatically. +datacap.executor.local.fetchSize=1000 +# Rows per batch flushed to the target via PreparedStatement.executeBatch(). +datacap.executor.local.batchSize=1000 + # ------ Apache Seatunnel for Flink ------ # # datacap.executor.data= # datacap.executor.way=LOCAL diff --git a/configure/etc/conf/application.properties b/configure/etc/conf/application.properties index 942fdafc57..a31d3d1925 100644 --- a/configure/etc/conf/application.properties +++ b/configure/etc/conf/application.properties @@ -48,6 +48,12 @@ datacap.executor.engine=SPARK datacap.executor.startScript=start-seatunnel-spark-connector-v2.sh datacap.executor.seatunnel.home=/opt/lib/seatunnel +# ------ Local Executor (in-process JDBC pipeline) ------ # +# JDBC fetchSize for streaming reads from the source. MySQL / MariaDB override to MIN_VALUE automatically. +datacap.executor.local.fetchSize=1000 +# Rows per batch flushed to the target via PreparedStatement.executeBatch(). +datacap.executor.local.batchSize=1000 + # ------ Apache Seatunnel for Flink ------ # # datacap.executor.data= # datacap.executor.way=LOCAL diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java index dd5be2df34..1f02a9018b 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java @@ -1044,6 +1044,8 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut String.join(File.separator, "dataset", entity.getExecutor().toLowerCase(), taskName) ); + int fetchSize = Integer.parseInt(environment.getProperty("datacap.executor.local.fetchSize", "1000")); + int batchSize = Integer.parseInt(environment.getProperty("datacap.executor.local.batchSize", "1000")); ExecutorRequest request = new ExecutorRequest( taskName, entity.getUser().getUsername(), @@ -1057,7 +1059,9 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut RunMode.valueOf(requireNonNull(environment.getProperty("datacap.executor.mode"))), environment.getProperty("datacap.executor.startScript"), RunEngine.valueOf(requireNonNull(environment.getProperty("datacap.executor.engine"))), - null + null, + fetchSize, + batchSize ); history.setState(RunState.RUNNING); diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/PluginService.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/PluginService.java index d9617bc27d..a93b10ce0c 100644 --- a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/PluginService.java +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/PluginService.java @@ -6,9 +6,12 @@ import com.google.common.collect.Sets; import io.edurt.datacap.plugin.Service; import io.edurt.datacap.spi.adapter.Adapter; +import io.edurt.datacap.spi.adapter.BatchWriter; import io.edurt.datacap.spi.adapter.HttpAdapter; import io.edurt.datacap.spi.adapter.JdbcAdapter; +import io.edurt.datacap.spi.adapter.JdbcStreamAdapter; import io.edurt.datacap.spi.adapter.NativeAdapter; +import io.edurt.datacap.spi.adapter.RowCallback; import io.edurt.datacap.spi.connection.Connection; import io.edurt.datacap.spi.connection.JdbcConnection; import io.edurt.datacap.spi.generator.DataType; @@ -180,6 +183,51 @@ default Response execute(Configure configure, String content) return this.execute(content); } + /** + * 当前插件是否支持流式读写 + * Whether the plugin supports streaming read / batched write. + * Default: only JDBC plugins. Override to opt-in or opt-out. + */ + default boolean supportsStreaming() + { + return type().equals(PluginType.JDBC); + } + + /** + * 流式执行查询，按行回调，不在 JVM 内物化结果集。 + * Stream rows out of the source one at a time. The implementation must NOT materialize + * the full result set in heap. + */ + default void executeStream(Configure configure, String content, int fetchSize, RowCallback callback) + { + if (type().equals(PluginType.JDBC)) { + JdbcStreamAdapter.executeStream(this, configure, content, fetchSize, callback); + } + else { + throw new UnsupportedOperationException("Streaming read is not supported for plugin " + this.name()); + } + } + + /** + * 打开批量写入会话，调用方按行 add 后由会话内部按 batchSize flush。 + * Open a batch-write session. Caller pushes rows; the writer flushes every batchSize rows + * and on close(). Must be closed by the caller. + */ + default BatchWriter openBatchWriter( + Configure configure, + String database, + String table, + java.util.List columns, + int batchSize) + { + if (type().equals(PluginType.JDBC)) { + return JdbcStreamAdapter.openBatchWriter(this, configure, database, table, columns, batchSize); + } + else { + throw new UnsupportedOperationException("Batch write is not supported for plugin " + this.name()); + } + } + /** * 获取数据库支持的引擎 * Get database supported engines diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/BatchWriter.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/BatchWriter.java new file mode 100644 index 0000000000..5878d4470c --- /dev/null +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/BatchWriter.java @@ -0,0 +1,14 @@ +package io.edurt.datacap.spi.adapter; + +import java.util.List; + +public interface BatchWriter + extends AutoCloseable +{ + void addRow(List row); + + long writtenCount(); + + @Override + void close(); +} diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcStreamAdapter.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcStreamAdapter.java new file mode 100644 index 0000000000..6fd8cabd07 --- /dev/null +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcStreamAdapter.java @@ -0,0 +1,306 @@ +package io.edurt.datacap.spi.adapter; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import io.edurt.datacap.spi.PluginService; +import io.edurt.datacap.spi.connection.JdbcConnection; +import io.edurt.datacap.spi.model.Configure; +import io.edurt.datacap.spi.model.Response; +import lombok.extern.slf4j.Slf4j; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +@Slf4j +@SuppressFBWarnings(value = {"OBL_UNSATISFIED_OBLIGATION", "OBL_UNSATISFIED_OBLIGATION_EXCEPTION_EDGE"}) +public final class JdbcStreamAdapter +{ + private static final int DEFAULT_FETCH_SIZE = 1000; + private static final int DEFAULT_BATCH_SIZE = 1000; + + private JdbcStreamAdapter() {} + + public static void executeStream( + PluginService plugin, + Configure configure, + String sql, + int fetchSize, + RowCallback callback) + { + int effectiveFetchSize = fetchSize > 0 ? fetchSize : DEFAULT_FETCH_SIZE; + JdbcConnection jdbcConnection = openConnection(plugin, configure); + try { + Connection connection = (Connection) jdbcConnection.getConnection(); + if (connection == null) { + throw new IllegalStateException("Open jdbc connection failed: " + + Optional.ofNullable(jdbcConnection.getResponse().getMessage()).orElse("unknown")); + } + + boolean restoreAutoCommit = false; + boolean originalAutoCommit = true; + try { + originalAutoCommit = connection.getAutoCommit(); + if (originalAutoCommit) { + connection.setAutoCommit(false); + restoreAutoCommit = true; + } + } + catch (SQLException ignore) { + // Some drivers (e.g. read-only HTTP based) do not support autoCommit toggling; just continue. + } + + try (Statement statement = connection.createStatement( + ResultSet.TYPE_FORWARD_ONLY, + ResultSet.CONCUR_READ_ONLY)) { + applyFetchSize(statement, effectiveFetchSize, configure); + try (ResultSet rs = statement.executeQuery(sql)) { + ResultSetMetaData metaData = rs.getMetaData(); + int columnCount = metaData.getColumnCount(); + List headers = new ArrayList<>(columnCount); + List types = new ArrayList<>(columnCount); + for (int i = 1; i <= columnCount; i++) { + String label = metaData.getColumnLabel(i); + headers.add(label != null && !label.isEmpty() ? label : metaData.getColumnName(i)); + types.add(metaData.getColumnTypeName(i)); + } + callback.onSchema(headers, types); + while (rs.next()) { + List row = new ArrayList<>(columnCount); + for (int i = 1; i <= columnCount; i++) { + row.add(rs.getObject(i)); + } + callback.onRow(row); + } + } + } + finally { + if (restoreAutoCommit) { + try { + connection.setAutoCommit(originalAutoCommit); + } + catch (SQLException ignore) { + } + } + } + } + catch (SQLException ex) { + throw new IllegalStateException("Stream read failed: " + ex.getMessage(), ex); + } + finally { + jdbcConnection.destroy(); + } + } + + public static BatchWriter openBatchWriter( + PluginService plugin, + Configure configure, + String database, + String table, + List columns, + int batchSize) + { + if (columns == null || columns.isEmpty()) { + throw new IllegalArgumentException("Batch writer requires at least one column"); + } + int effectiveBatchSize = batchSize > 0 ? batchSize : DEFAULT_BATCH_SIZE; + JdbcConnection jdbcConnection = openConnection(plugin, configure); + Connection connection = (Connection) jdbcConnection.getConnection(); + if (connection == null) { + jdbcConnection.destroy(); + throw new IllegalStateException("Open jdbc connection failed: " + + Optional.ofNullable(jdbcConnection.getResponse().getMessage()).orElse("unknown")); + } + return new JdbcBatchWriter(jdbcConnection, connection, database, table, columns, effectiveBatchSize); + } + + private static JdbcConnection openConnection(PluginService plugin, Configure configure) + { + Response response = new Response(); + configure.setDriver(plugin.driver()); + configure.setType(plugin.connectType()); + configure.setUrl(Optional.of(plugin.url(configure))); + return new JdbcConnection(configure, response); + } + + private static void applyFetchSize(Statement statement, int fetchSize, Configure configure) + { + try { + String type = configure.getType(); + // MySQL only streams when fetchSize is Integer.MIN_VALUE with TYPE_FORWARD_ONLY + CONCUR_READ_ONLY. + // The plugin's connectType is "datacap"; the original driver type is in the plugin itself, but the URL + // encodes the actual database. Detect by URL prefix. + String url = configure.getUrl().orElse(""); + if (url.startsWith("jdbc:mysql") || url.startsWith("jdbc:mariadb")) { + statement.setFetchSize(Integer.MIN_VALUE); + } + else { + statement.setFetchSize(fetchSize); + } + if (type != null && type.toLowerCase().contains("postgres")) { + // PostgreSQL also needs autoCommit=false for cursor-based fetch (handled by caller). + statement.setFetchSize(fetchSize); + } + } + catch (SQLException ex) { + log.warn("Set fetch size failed, falling back to driver default: {}", ex.getMessage()); + } + } + + private static final class JdbcBatchWriter + implements BatchWriter + { + private final JdbcConnection jdbcConnection; + private final Connection connection; + private final PreparedStatement statement; + private final int columnCount; + private final int batchSize; + private int pending; + private long written; + private boolean originalAutoCommit = true; + private boolean restoreAutoCommit; + + JdbcBatchWriter( + JdbcConnection jdbcConnection, + Connection connection, + String database, + String table, + List columns, + int batchSize) + { + this.jdbcConnection = jdbcConnection; + this.connection = connection; + this.columnCount = columns.size(); + this.batchSize = batchSize; + + try { + this.originalAutoCommit = connection.getAutoCommit(); + if (this.originalAutoCommit) { + connection.setAutoCommit(false); + this.restoreAutoCommit = true; + } + } + catch (SQLException ignore) { + } + + String sql = buildInsertTemplate(database, table, columns); + try { + this.statement = connection.prepareStatement(sql); + } + catch (SQLException ex) { + safeClose(); + throw new IllegalStateException("Prepare insert failed: " + ex.getMessage(), ex); + } + } + + @Override + public void addRow(List row) + { + if (row == null || row.size() != columnCount) { + throw new IllegalArgumentException( + "Row size " + (row == null ? -1 : row.size()) + " does not match column count " + columnCount); + } + try { + for (int i = 0; i < columnCount; i++) { + Object value = row.get(i); + if (value == null) { + statement.setObject(i + 1, null); + } + else { + statement.setObject(i + 1, value); + } + } + statement.addBatch(); + pending++; + if (pending >= batchSize) { + flush(); + } + } + catch (SQLException ex) { + throw new IllegalStateException("Add batch row failed: " + ex.getMessage(), ex); + } + } + + @Override + public long writtenCount() + { + return written; + } + + @Override + public void close() + { + try { + if (pending > 0) { + flush(); + } + } + catch (SQLException ex) { + throw new IllegalStateException("Flush final batch failed: " + ex.getMessage(), ex); + } + finally { + safeClose(); + } + } + + private void flush() + throws SQLException + { + statement.executeBatch(); + connection.commit(); + written += pending; + pending = 0; + statement.clearBatch(); + } + + private void safeClose() + { + try { + if (statement != null) { + statement.close(); + } + } + catch (SQLException ex) { + log.warn("Close prepared statement failed: {}", ex.getMessage()); + } + try { + if (restoreAutoCommit) { + connection.setAutoCommit(originalAutoCommit); + } + } + catch (SQLException ignore) { + } + jdbcConnection.destroy(); + } + } + + private static String buildInsertTemplate(String database, String table, List columns) + { + StringBuilder sb = new StringBuilder(); + sb.append("INSERT INTO "); + if (database != null && !database.isEmpty()) { + sb.append('`').append(database).append("`."); + } + sb.append('`').append(table).append("` ("); + for (int i = 0; i < columns.size(); i++) { + if (i > 0) { + sb.append(", "); + } + sb.append('`').append(columns.get(i)).append('`'); + } + sb.append(") VALUES ("); + for (int i = 0; i < columns.size(); i++) { + if (i > 0) { + sb.append(", "); + } + sb.append('?'); + } + sb.append(')'); + return sb.toString(); + } +} diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/RowCallback.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/RowCallback.java new file mode 100644 index 0000000000..e8ac5aa26a --- /dev/null +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/RowCallback.java @@ -0,0 +1,18 @@ +package io.edurt.datacap.spi.adapter; + +import java.util.List; + +public interface RowCallback +{ + /** + * 在第一行之前回调一次，传入列元数据。默认空实现。 + * Called once before any row, carrying the column metadata of the source result. + */ + default void onSchema(List headers, List types) {} + + /** + * 每一行回调一次。 + * Called per row. Values are aligned to {@code headers} from {@link #onSchema}. + */ + void onRow(List row); +} diff --git a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt index 7ee009dc73..9c7da99573 100644 --- a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt +++ b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt @@ -1,7 +1,8 @@ package io.edurt.datacap.executor.local +import com.fasterxml.jackson.databind.JsonNode import com.fasterxml.jackson.databind.node.ObjectNode -import com.google.common.collect.Lists +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings import io.edurt.datacap.common.sql.SqlBuilder import io.edurt.datacap.common.sql.configure.SqlBody import io.edurt.datacap.common.sql.configure.SqlColumn @@ -10,10 +11,18 @@ import io.edurt.datacap.executor.ExecutorService import io.edurt.datacap.executor.common.RunState import io.edurt.datacap.executor.configure.ExecutorRequest import io.edurt.datacap.executor.configure.ExecutorResponse -import org.apache.commons.lang.StringEscapeUtils +import io.edurt.datacap.executor.configure.OriginColumn +import io.edurt.datacap.spi.PluginService +import io.edurt.datacap.spi.adapter.BatchWriter +import io.edurt.datacap.spi.adapter.RowCallback +import io.edurt.datacap.spi.model.Configure +import org.slf4j.LoggerFactory +@SuppressFBWarnings(value = ["BC_BAD_CAST_TO_ABSTRACT_COLLECTION"]) class LocalExecutorService : ExecutorService { + private val log = LoggerFactory.getLogger(LocalExecutorService::class.java) + override fun start(request: ExecutorRequest): ExecutorResponse { val response = ExecutorResponse() @@ -21,73 +30,49 @@ class LocalExecutorService : ExecutorService { val input = request.input val output = request.output - val inputPlugin = input.plugin - if (inputPlugin != null) + val inputPlugin = input.plugin ?: throw IllegalArgumentException("Input plugin is null") + val outputPlugin = output.plugin ?: throw IllegalArgumentException("Output plugin is null") + val inputConfigure = input.originConfigure ?: throw IllegalArgumentException("Input configure is null") + val outputConfigure = output.originConfigure ?: throw IllegalArgumentException("Output configure is null") + val database = output.database ?: throw IllegalArgumentException("Output database is null") + val table = output.table ?: throw IllegalArgumentException("Output table is null") + val query = input.query ?: throw IllegalArgumentException("Input query is null") + + val originColumns = input.originColumns.toList() + if (originColumns.isEmpty()) { - val inputResult = inputPlugin.execute(input.originConfigure, input.query) + throw IllegalArgumentException("Input column mapping is empty") + } + val targetColumns = originColumns.map { it.name } + val sourceKeys = originColumns.map { it.original } - if (inputResult.isSuccessful) - { - val fullInsertStatement = Lists.newArrayList() - response.count = inputResult.columns.size - inputResult.columns.forEach { it -> - val columns = Lists.newArrayList() - val objectNode: ObjectNode = it as ObjectNode - input.originColumns.forEach { value -> - columns.add( - SqlColumn.builder() - .column(String.format("`%s`", value.name)) - .value(String.format("'%s'", StringEscapeUtils.escapeSql(objectNode.get(value.original).asText()))) - .build() - ) - } - val body = SqlBody.builder() - .type(SqlType.INSERT) - .database(input.database) - .table(input.table) - .columns(columns) - .build() - fullInsertStatement.add(SqlBuilder(body).sql) - } + val fetchSize = if (request.fetchSize > 0) request.fetchSize else DEFAULT_FETCH_SIZE + val batchSize = if (request.batchSize > 0) request.batchSize else DEFAULT_BATCH_SIZE - if (fullInsertStatement.isNotEmpty()) - { - val outputPlugin = output.plugin - if (outputPlugin != null) - { - val outputResult = outputPlugin.execute(output.originConfigure, fullInsertStatement.joinToString("\n\n")) - if (! outputResult.isSuccessful) - { - throw RuntimeException(outputResult.message) - } - else - { - response.successful = true - response.state = RunState.SUCCESS - } - } - else - { - throw RuntimeException("Output plugin is null") - } - } - else - { - throw RuntimeException("No data to insert") - } - } - else - { - throw RuntimeException(inputResult.message) - } + val written = if (inputPlugin.supportsStreaming() && outputPlugin.supportsStreaming()) + { + runStreaming( + inputPlugin, inputConfigure, query, fetchSize, + outputPlugin, outputConfigure, database, table, + targetColumns, sourceKeys, batchSize + ) } else { - throw RuntimeException("Input plugin is null") + runLegacy( + inputPlugin, inputConfigure, query, + outputPlugin, outputConfigure, database, table, + originColumns, batchSize + ) } + + response.count = if (written > Int.MAX_VALUE.toLong()) Int.MAX_VALUE else written.toInt() + response.successful = true + response.state = RunState.SUCCESS } catch (ex: Exception) { + log.error("Local executor failed", ex) response.successful = false response.state = RunState.FAILURE response.message = ex.message @@ -97,6 +82,225 @@ class LocalExecutorService : ExecutorService override fun stop(request: ExecutorRequest): ExecutorResponse { - TODO("Not yet implemented") + return ExecutorResponse(false, true, RunState.SUCCESS, null) + } + + /** + * 流式路径：源端 fetchSize 拉取，目标端 PreparedStatement 批量写。 + * 全程不在 JVM 内物化整个结果集。 + */ + private fun runStreaming( + inputPlugin: PluginService, + inputConfigure: Configure, + query: String, + fetchSize: Int, + outputPlugin: PluginService, + outputConfigure: Configure, + database: String, + table: String, + targetColumns: List, + sourceKeys: List, + batchSize: Int + ): Long + { + var written = 0L + val writer: BatchWriter = outputPlugin.openBatchWriter( + outputConfigure, database, table, targetColumns, batchSize + ) + writer.use { writer -> + val indexByHeader = HashMap() + inputPlugin.executeStream(inputConfigure, query, fetchSize, object : RowCallback + { + override fun onSchema(headers: List, types: List) + { + indexByHeader.clear() + headers.forEachIndexed { i, h -> indexByHeader[h.lowercase()] = i } + } + + override fun onRow(row: List) + { + val projected = ArrayList(sourceKeys.size) + for (key in sourceKeys) + { + val idx = indexByHeader[key.lowercase()] + ?: throw IllegalStateException("Source column '$key' not found in query result") + projected.add(row[idx]) + } + writer.addRow(projected) + written ++ + } + }) + } + return written + } + + /** + * 回退路径：源或汇任一不支持流式（如 HTTP / Native 插件）。仍然使用旧的全量读取， + * 但目标端按 batch 切片提交，避免一次性拼接巨大 SQL 字符串；同时修复 NULL、类型、转义问题。 + */ + private fun runLegacy( + inputPlugin: PluginService, + inputConfigure: Configure, + query: String, + outputPlugin: PluginService, + outputConfigure: Configure, + database: String, + table: String, + originColumns: List, + batchSize: Int + ): Long + { + val inputResult = inputPlugin.execute(inputConfigure, query) + if (inputResult.isSuccessful != true) + { + throw RuntimeException(inputResult.message ?: "Input plugin failed") + } + val rows = inputResult.columns ?: return 0L + + // 目标端能流式：用 BatchWriter 安全 + 节省内存 + if (outputPlugin.supportsStreaming()) + { + val targetColumns = originColumns.map { it.name } + val sourceKeys = originColumns.map { it.original } + var written = 0L + val writer = outputPlugin.openBatchWriter( + outputConfigure, database, table, targetColumns, batchSize + ) + writer.use { writer -> + for (item in rows) + { + val node = item as? ObjectNode ?: continue + val projected = ArrayList(sourceKeys.size) + for (key in sourceKeys) + { + projected.add(jsonNodeToJdbcValue(node.get(key))) + } + writer.addRow(projected) + written ++ + } + } + return written + } + + // 双端都不支持流式：用 INSERT 字符串，但按 batch 提交，不再拼一个巨大字符串 + var written = 0L + val batch = ArrayList(batchSize) + for (item in rows) + { + val node = item as? ObjectNode ?: continue + val sqlColumns = ArrayList(originColumns.size) + for (col in originColumns) + { + sqlColumns.add( + SqlColumn.builder() + .column("`${col.name}`") + .value(formatSqlLiteral(node.get(col.original))) + .build() + ) + } + val body = SqlBody.builder() + .type(SqlType.INSERT) + .database(database) + .table(table) + .columns(sqlColumns) + .build() + batch.add(SqlBuilder(body).sql) + if (batch.size >= batchSize) + { + flushLegacyBatch(outputPlugin, outputConfigure, batch) + written += batch.size + batch.clear() + } + } + if (batch.isNotEmpty()) + { + flushLegacyBatch(outputPlugin, outputConfigure, batch) + written += batch.size + batch.clear() + } + return written + } + + private fun flushLegacyBatch(plugin: PluginService, configure: Configure, batch: List) + { + val joined = batch.joinToString("\n") + val result = plugin.execute(configure, joined) + if (result.isSuccessful != true) + { + throw RuntimeException(result.message ?: "Output plugin failed") + } + } + + /** + * 把 JsonNode 转成 JDBC 可识别的真实类型；保留 NULL 语义。 + */ + private fun jsonNodeToJdbcValue(node: JsonNode?): Any? + { + if (node == null || node.isNull) return null + return when + { + node.isBoolean -> node.asBoolean() + node.isInt -> node.asInt() + node.isLong -> node.asLong() + node.isBigInteger -> node.bigIntegerValue() + node.isFloat || node.isDouble -> node.asDouble() + node.isBigDecimal -> node.decimalValue() + node.isBinary -> + { + try + { + node.binaryValue() + } + catch (e: Exception) + { + node.asText() + } + } + + else -> node.asText() + } + } + + /** + * 旧路径下需要拼 SQL 字符串：按类型生成字面量，正确处理 NULL / 数字 / 布尔 / 字符串转义。 + * 仅在双端均不支持流式（HTTP / Native 输出插件）时使用。 + */ + private fun formatSqlLiteral(node: JsonNode?): String + { + if (node == null || node.isNull) return "NULL" + return when + { + node.isBoolean -> if (node.asBoolean()) "TRUE" else "FALSE" + node.isIntegralNumber -> node.asLong().toString() + node.isFloatingPointNumber || node.isBigDecimal -> node.asText() + else -> "'${escapeSqlString(node.asText())}'" + } + } + + private fun escapeSqlString(s: String): String + { + if (s.isEmpty()) return s + val sb = StringBuilder(s.length + 4) + for (c in s) + { + when (c) + { + '\'' -> sb.append("''") + '\\' -> sb.append("\\\\") + '\u0000' -> + { + // SQL 不允许 NUL 字符，跳过以避免协议层错误 + } + + else -> sb.append(c) + } + } + return sb.toString() + } + + companion object + { + private const val DEFAULT_FETCH_SIZE = 1000 + private const val DEFAULT_BATCH_SIZE = 1000 } } diff --git a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorRequest.kt b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorRequest.kt index 81d58ec1fe..b88b4408a9 100644 --- a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorRequest.kt +++ b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorRequest.kt @@ -7,7 +7,7 @@ import io.edurt.datacap.executor.common.RunWay import io.edurt.datacap.plugin.PluginManager @SuppressFBWarnings(value = ["EI_EXPOSE_REP", "EI_EXPOSE_REP2"]) -data class ExecutorRequest( +data class ExecutorRequest @JvmOverloads constructor( var taskName: String, var userName: String, var input: ExecutorConfigure, @@ -20,7 +20,9 @@ data class ExecutorRequest( var runMode: RunMode = RunMode.CLIENT, var startScript: String? = null, var runEngine: RunEngine = RunEngine.SPARK, - var transform: ExecutorConfigure? = null + var transform: ExecutorConfigure? = null, + var fetchSize: Int = 1000, + var batchSize: Int = 1000 ) { constructor( From 73856dc0e9bbf13c764c74c67faba07fd7306c3d Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Wed, 20 May 2026 15:19:04 +0800 Subject: [PATCH 03/23] =?UTF-8?q?fix(executor-local):=20=E4=BB=8E=20input?= =?UTF-8?q?=20=E8=AF=BB=E5=8F=96=E7=9B=AE=E6=A0=87=E5=BA=93=20/=20?= =?UTF-8?q?=E8=A1=A8=EF=BC=8C=E4=B8=8E=20DataSetServiceImpl=20=E7=BA=A6?= =?UTF-8?q?=E5=AE=9A=E5=AF=B9=E9=BD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../io/edurt/datacap/executor/local/LocalExecutorService.kt | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt index 9c7da99573..45e20f8b58 100644 --- a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt +++ b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt @@ -34,8 +34,9 @@ class LocalExecutorService : ExecutorService val outputPlugin = output.plugin ?: throw IllegalArgumentException("Output plugin is null") val inputConfigure = input.originConfigure ?: throw IllegalArgumentException("Input configure is null") val outputConfigure = output.originConfigure ?: throw IllegalArgumentException("Output configure is null") - val database = output.database ?: throw IllegalArgumentException("Output database is null") - val table = output.table ?: throw IllegalArgumentException("Output table is null") + // DataCap 约定：目标库 / 目标表 由调用方放在 input 上（见 DataSetServiceImpl） + val database = input.database ?: throw IllegalArgumentException("Target database is null") + val table = input.table ?: throw IllegalArgumentException("Target table is null") val query = input.query ?: throw IllegalArgumentException("Input query is null") val originColumns = input.originColumns.toList() From 90d77e6b08dd38bac0e92b33b70ceeb7c2fd3932 Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Wed, 20 May 2026 18:19:05 +0800 Subject: [PATCH 04/23] =?UTF-8?q?feat(executor-local):=20=E8=BE=93?= =?UTF-8?q?=E5=87=BA=E7=8B=AC=E7=AB=8B=E4=BB=BB=E5=8A=A1=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E5=B9=B6=E5=8A=A0=E8=BF=9B=E5=BA=A6=E7=BB=9F=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 引入 datacap-logger，按 SeaTunnel 同款方式在 workHome 下生成 {taskName}.log - 每 10000 行打印一次进度（已读 / 已 commit / 耗时 / 行/秒） - 同步开始 / 结束 / 异常均输出到该任务专属日志 - pom 加 maven-dependency-plugin:copy-dependencies，确保 IDE 开发模式下 PluginClassLoader 能在 target/dependencies 找到依赖 jar --- executor/datacap-executor-local/pom.xml | 23 +++++ .../executor/local/LocalExecutorService.kt | 94 +++++++++++++++++-- 2 files changed, 111 insertions(+), 6 deletions(-) diff --git a/executor/datacap-executor-local/pom.xml b/executor/datacap-executor-local/pom.xml index 53d18ae770..ed1b1fe231 100644 --- a/executor/datacap-executor-local/pom.xml +++ b/executor/datacap-executor-local/pom.xml @@ -24,6 +24,11 @@ kotlin-reflect provided + + io.edurt.datacap + datacap-logger + ${project.version} + @@ -47,6 +52,24 @@ + + org.apache.maven.plugins + maven-dependency-plugin + + + copy-dependencies + package + + copy-dependencies + + + ${project.build.directory}/dependencies + provided + runtime + + + + org.jetbrains.dokka dokka-maven-plugin diff --git a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt index 45e20f8b58..42f3959ec1 100644 --- a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt +++ b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt @@ -12,13 +12,16 @@ import io.edurt.datacap.executor.common.RunState import io.edurt.datacap.executor.configure.ExecutorRequest import io.edurt.datacap.executor.configure.ExecutorResponse import io.edurt.datacap.executor.configure.OriginColumn +import io.edurt.datacap.lib.logger.LoggerExecutor +import io.edurt.datacap.lib.logger.logback.LogbackExecutor import io.edurt.datacap.spi.PluginService import io.edurt.datacap.spi.adapter.BatchWriter import io.edurt.datacap.spi.adapter.RowCallback import io.edurt.datacap.spi.model.Configure +import org.slf4j.Logger import org.slf4j.LoggerFactory -@SuppressFBWarnings(value = ["BC_BAD_CAST_TO_ABSTRACT_COLLECTION"]) +@SuppressFBWarnings(value = ["BC_BAD_CAST_TO_ABSTRACT_COLLECTION", "RV_RETURN_VALUE_IGNORED_BAD_PRACTICE"]) class LocalExecutorService : ExecutorService { private val log = LoggerFactory.getLogger(LocalExecutorService::class.java) @@ -26,8 +29,12 @@ class LocalExecutorService : ExecutorService override fun start(request: ExecutorRequest): ExecutorResponse { val response = ExecutorResponse() + val loggerExecutor: LoggerExecutor<*>? = newTaskLogger(request) + val taskLog: Logger = loggerExecutor?.getLogger() ?: log try { + taskLog.info("Local executor task starting: task={} user={}", request.taskName, request.userName) + val input = request.input val output = request.output val inputPlugin = input.plugin ?: throw IllegalArgumentException("Input plugin is null") @@ -50,12 +57,20 @@ class LocalExecutorService : ExecutorService val fetchSize = if (request.fetchSize > 0) request.fetchSize else DEFAULT_FETCH_SIZE val batchSize = if (request.batchSize > 0) request.batchSize else DEFAULT_BATCH_SIZE + taskLog.info( + "Resolved input plugin={} output plugin={} streaming={}/{} fetchSize={} batchSize={}", + inputPlugin.name(), outputPlugin.name(), + inputPlugin.supportsStreaming(), outputPlugin.supportsStreaming(), + fetchSize, batchSize + ) + taskLog.info("Source query: {}", query) + val written = if (inputPlugin.supportsStreaming() && outputPlugin.supportsStreaming()) { runStreaming( inputPlugin, inputConfigure, query, fetchSize, outputPlugin, outputConfigure, database, table, - targetColumns, sourceKeys, batchSize + targetColumns, sourceKeys, batchSize, taskLog ) } else @@ -63,24 +78,53 @@ class LocalExecutorService : ExecutorService runLegacy( inputPlugin, inputConfigure, query, outputPlugin, outputConfigure, database, table, - originColumns, batchSize + originColumns, batchSize, taskLog ) } response.count = if (written > Int.MAX_VALUE.toLong()) Int.MAX_VALUE else written.toInt() response.successful = true response.state = RunState.SUCCESS + taskLog.info("Local executor task completed: rows={} state=SUCCESS", written) } catch (ex: Exception) { - log.error("Local executor failed", ex) + taskLog.error("Local executor failed", ex) response.successful = false response.state = RunState.FAILURE response.message = ex.message } + finally + { + loggerExecutor?.destroy() + } return response } + private fun newTaskLogger(request: ExecutorRequest): LoggerExecutor<*>? + { + val workHome = request.workHome + val taskName = request.taskName + if (workHome.isNullOrBlank() || taskName.isBlank()) + { + return null + } + return try + { + val dir = java.io.File(workHome) + if (!dir.exists()) + { + dir.mkdirs() + } + LogbackExecutor(workHome, "$taskName.log") + } + catch (ex: Exception) + { + log.warn("Create task logger at {} failed: {}", workHome, ex.message) + null + } + } + override fun stop(request: ExecutorRequest): ExecutorResponse { return ExecutorResponse(false, true, RunState.SUCCESS, null) @@ -101,9 +145,15 @@ class LocalExecutorService : ExecutorService table: String, targetColumns: List, sourceKeys: List, - batchSize: Int + batchSize: Int, + taskLog: Logger ): Long { + taskLog.info( + "Streaming sync start: target=`{}`.`{}` columns={} fetchSize={} batchSize={}", + database, table, targetColumns.size, fetchSize, batchSize + ) + val startNanos = System.nanoTime() var written = 0L val writer: BatchWriter = outputPlugin.openBatchWriter( outputConfigure, database, table, targetColumns, batchSize @@ -116,6 +166,7 @@ class LocalExecutorService : ExecutorService { indexByHeader.clear() headers.forEachIndexed { i, h -> indexByHeader[h.lowercase()] = i } + taskLog.info("Streaming sync: source returned headers={}", headers) } override fun onRow(row: List) @@ -129,9 +180,23 @@ class LocalExecutorService : ExecutorService } writer.addRow(projected) written ++ + if (written % PROGRESS_INTERVAL == 0L) + { + val seconds = (System.nanoTime() - startNanos) / 1_000_000_000.0 + val rps = if (seconds > 0) (written / seconds).toLong() else 0L + taskLog.info( + "Streaming sync progress: read={} committed={} elapsed={}s rps={}", + written, writer.writtenCount(), "%.1f".format(seconds), rps + ) + } } }) } + val totalSeconds = (System.nanoTime() - startNanos) / 1_000_000_000.0 + taskLog.info( + "Streaming sync done: rows={} committed={} elapsed={}s", + written, writer.writtenCount(), "%.1f".format(totalSeconds) + ) return written } @@ -148,15 +213,19 @@ class LocalExecutorService : ExecutorService database: String, table: String, originColumns: List, - batchSize: Int + batchSize: Int, + taskLog: Logger ): Long { + taskLog.info("Legacy sync start: target=`{}`.`{}` batchSize={}", database, table, batchSize) + val startNanos = System.nanoTime() val inputResult = inputPlugin.execute(inputConfigure, query) if (inputResult.isSuccessful != true) { throw RuntimeException(inputResult.message ?: "Input plugin failed") } val rows = inputResult.columns ?: return 0L + taskLog.info("Legacy sync: source materialized {} rows", rows.size) // 目标端能流式：用 BatchWriter 安全 + 节省内存 if (outputPlugin.supportsStreaming()) @@ -178,8 +247,14 @@ class LocalExecutorService : ExecutorService } writer.addRow(projected) written ++ + if (written % PROGRESS_INTERVAL == 0L) + { + taskLog.info("Legacy sync progress: written={} committed={}", written, writer.writtenCount()) + } } } + val totalSeconds = (System.nanoTime() - startNanos) / 1_000_000_000.0 + taskLog.info("Legacy sync done: rows={} committed={} elapsed={}s", written, writer.writtenCount(), "%.1f".format(totalSeconds)) return written } @@ -211,6 +286,10 @@ class LocalExecutorService : ExecutorService flushLegacyBatch(outputPlugin, outputConfigure, batch) written += batch.size batch.clear() + if (written % PROGRESS_INTERVAL == 0L) + { + taskLog.info("Legacy sync progress (sql batch): written={}", written) + } } } if (batch.isNotEmpty()) @@ -219,6 +298,8 @@ class LocalExecutorService : ExecutorService written += batch.size batch.clear() } + val totalSeconds = (System.nanoTime() - startNanos) / 1_000_000_000.0 + taskLog.info("Legacy sync done: rows={} elapsed={}s", written, "%.1f".format(totalSeconds)) return written } @@ -303,5 +384,6 @@ class LocalExecutorService : ExecutorService { private const val DEFAULT_FETCH_SIZE = 1000 private const val DEFAULT_BATCH_SIZE = 1000 + private const val PROGRESS_INTERVAL = 10_000L } } From 2bc5dd926ad7597747fc514c6e681819ecbceddc Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Wed, 20 May 2026 18:29:42 +0800 Subject: [PATCH 05/23] =?UTF-8?q?feat(executor-spi):=20=E5=A2=9E=E5=8A=A0?= =?UTF-8?q?=20ExecutorService.name()=20=E5=B9=B6=E7=94=A8=E4=BA=8E=20workH?= =?UTF-8?q?ome=20=E8=B7=AF=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ExecutorService SPI 加 default name() 方法，默认去掉类名末尾的 "ExecutorService"，保留大小写 - DataSetServiceImpl 用 executorService.name().toLowerCase() 作为 workHome 中的 executor 类型段 - 路径格式调整为 dataset/executor/{executor type}/{taskName} --- .../service/service/impl/DataSetServiceImpl.java | 2 +- .../io/edurt/datacap/executor/ExecutorService.kt | 11 +++++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java index 1f02a9018b..254fde1c0d 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java @@ -1041,7 +1041,7 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut String workHome = FolderUtils.getWorkHome( initializerConfigure.getDataHome(), entity.getUser().getUsername(), - String.join(File.separator, "dataset", entity.getExecutor().toLowerCase(), taskName) + String.join(File.separator, "dataset", "executor", executorService.name().toLowerCase(), taskName) ); int fetchSize = Integer.parseInt(environment.getProperty("datacap.executor.local.fetchSize", "1000")); diff --git a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/ExecutorService.kt b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/ExecutorService.kt index da17898fe9..298992e96e 100644 --- a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/ExecutorService.kt +++ b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/ExecutorService.kt @@ -9,4 +9,15 @@ interface ExecutorService : Service fun start(request: ExecutorRequest): ExecutorResponse fun stop(request: ExecutorRequest): ExecutorResponse + + /** + * 当前 Executor 的短名。默认实现：去掉类名末尾的 "ExecutorService"。 + * 例如 LocalExecutorService -> "Local"，SeatunnelExecutorService -> "Seatunnel"。 + * 大小写转换由调用方按业务需要处理。 + * + * Short name of this executor. Default: strip "ExecutorService" suffix from the class name. + * e.g. LocalExecutorService -> "Local", SeatunnelExecutorService -> "Seatunnel". + * Case conversion is the caller's responsibility. + */ + fun name(): String = this::class.java.simpleName.removeSuffix("ExecutorService") } From 28777f8f4e1ed3f63de1fea0ddd00de6a8699e3f Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Wed, 20 May 2026 19:57:46 +0800 Subject: [PATCH 06/23] refactor(executor): unify executor identifier via SPI ExecutorPlugin base - introduce ExecutorPlugin base in datacap-executor-spi: auto-derives getName() by stripping the "Executor" suffix from the subclass name and fixes getType() to EXECUTOR, so each plugin needs zero boilerplate - rename SeatunnelExecutorPlugin -> SeatunnelExecutor, LocalExecutorPlugin -> LocalExecutor; both now extend ExecutorPlugin with empty bodies - drop the redundant ExecutorService.name() default; Plugin.getName() is the single source of truth for the executor short name - WorkflowServiceImpl / DataSetServiceImpl: derive the property key from plugin.getName().toLowerCase() so datacap.executor..home actually resolves; previously the lookup used the long form and silently returned null, producing "null/bin/start-seatunnel-..." commands - UI: align hardcoded executor identifiers ('SeatunnelExecutor' -> 'Seatunnel', 'LocalExecutor' -> 'Local') in FlowEditor, DatasetInfo and WorkflowInfo - configure/etc/conf/executor/category.yaml and datacap.sql default value switched to the short form - add 2026.0.0/schema.sql migration to rewrite existing datacap_dataset and datacap_workflow rows to the short form - enable -Xjvm-default=all in the root kotlin-maven-plugin so Kotlin interface defaults compile as JVM 8 default methods and Java implementations inherit them without AbstractMethodError - update executor pluginName tests to the short form --- configure/etc/conf/executor/category.yaml | 6 ++--- configure/schema/2026.0.0/schema.sql | 25 +++++++++++++++++++ configure/schema/datacap.sql | 2 +- .../service/impl/DataSetServiceImpl.java | 5 ++-- .../service/impl/WorkflowServiceImpl.java | 2 +- .../components/editor/flow/FlowEditor.vue | 2 +- .../views/pages/admin/dataset/DatasetInfo.vue | 2 +- .../pages/admin/wofkflow/WorkflowInfo.vue | 4 +-- .../datacap/executor/local/LocalExecutor.kt | 5 ++++ .../executor/local/LocalExecutorPlugin.kt | 12 --------- .../services/io.edurt.datacap.plugin.Plugin | 2 +- .../executor/seatunnel/SeatunnelExecutor.java | 8 ++++++ .../seatunnel/SeatunnelExecutorPlugin.java | 14 ----------- .../services/io.edurt.datacap.plugin.Plugin | 2 +- .../edurt/datacap/executor/ExecutorPlugin.kt | 22 ++++++++++++++++ .../edurt/datacap/executor/ExecutorService.kt | 11 -------- pom.xml | 3 +++ .../test/local/LocalExecutorPluginTest.kt | 2 +- .../test/local/LocalExecutorServiceTest.kt | 2 +- .../seatunnel/SeatunnelExecutorPluginTest.kt | 2 +- 20 files changed, 80 insertions(+), 53 deletions(-) create mode 100644 configure/schema/2026.0.0/schema.sql create mode 100644 executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutor.kt delete mode 100644 executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorPlugin.kt create mode 100644 executor/datacap-executor-seatunnel/src/main/java/io/edurt/datacap/executor/seatunnel/SeatunnelExecutor.java delete mode 100644 executor/datacap-executor-seatunnel/src/main/java/io/edurt/datacap/executor/seatunnel/SeatunnelExecutorPlugin.java create mode 100644 executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/ExecutorPlugin.kt diff --git a/configure/etc/conf/executor/category.yaml b/configure/etc/conf/executor/category.yaml index 8c99f9e6cc..48984be673 100644 --- a/configure/etc/conf/executor/category.yaml +++ b/configure/etc/conf/executor/category.yaml @@ -1,12 +1,12 @@ - label: Input value: source supportExecutors: - - SeatunnelExecutor + - Seatunnel - label: Transform value: transform supportExecutors: - - SeatunnelExecutor + - Seatunnel - label: Output value: sink supportExecutors: - - SeatunnelExecutor + - Seatunnel diff --git a/configure/schema/2026.0.0/schema.sql b/configure/schema/2026.0.0/schema.sql new file mode 100644 index 0000000000..14d6cf7de2 --- /dev/null +++ b/configure/schema/2026.0.0/schema.sql @@ -0,0 +1,25 @@ +USE +`datacap`; + +-- Executor 命名统一：去掉冗余的 "Executor" 后缀，与 SPI ExecutorService.name() / ExecutorPlugin.getName() 对齐。 +-- 例如 'SeatunnelExecutor' -> 'Seatunnel'，'LocalExecutor' -> 'Local'。 +-- Unify executor identifier with SPI's name(): strip the redundant "Executor" suffix. + +UPDATE `datacap_dataset` +SET `executor` = 'Local' +WHERE `executor` = 'LocalExecutor'; + +UPDATE `datacap_dataset` +SET `executor` = 'Seatunnel' +WHERE `executor` = 'SeatunnelExecutor'; + +ALTER TABLE `datacap_dataset` + ALTER COLUMN `executor` SET DEFAULT 'Local'; + +UPDATE `datacap_workflow` +SET `executor` = 'Local' +WHERE `executor` = 'LocalExecutor'; + +UPDATE `datacap_workflow` +SET `executor` = 'Seatunnel' +WHERE `executor` = 'SeatunnelExecutor'; diff --git a/configure/schema/datacap.sql b/configure/schema/datacap.sql index 9f979e224b..e076248b6b 100644 --- a/configure/schema/datacap.sql +++ b/configure/schema/datacap.sql @@ -164,7 +164,7 @@ CREATE TABLE `datacap_dataset` ( `code` varchar(100) DEFAULT (uuid()), `column_mode` varchar(100) DEFAULT 'DIMENSION', `scheduler` varchar(100) DEFAULT 'LocalScheduler', - `executor` varchar(100) DEFAULT 'LocalExecutor', + `executor` varchar(100) DEFAULT 'Local', `total_rows` bigint DEFAULT '0', `total_size` varchar(100) DEFAULT NULL, `life_cycle` bigint DEFAULT '0', diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java index 254fde1c0d..a47c82e66e 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java @@ -1038,10 +1038,11 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut ); String taskName = DateFormatUtils.format(System.currentTimeMillis(), "yyyyMMddHHmmssSSS"); + String executorKey = executor.getName().toLowerCase(); String workHome = FolderUtils.getWorkHome( initializerConfigure.getDataHome(), entity.getUser().getUsername(), - String.join(File.separator, "dataset", "executor", executorService.name().toLowerCase(), taskName) + String.join(File.separator, "dataset", "executor", executorKey, taskName) ); int fetchSize = Integer.parseInt(environment.getProperty("datacap.executor.local.fetchSize", "1000")); @@ -1051,7 +1052,7 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut entity.getUser().getUsername(), input, output, - environment.getProperty(String.format("datacap.executor.%s.home", entity.getExecutor().toLowerCase())), + environment.getProperty(String.format("datacap.executor.%s.home", executorKey)), workHome, this.pluginManager, 600, diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/WorkflowServiceImpl.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/WorkflowServiceImpl.java index f5143c6f67..6cbeb5c194 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/WorkflowServiceImpl.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/WorkflowServiceImpl.java @@ -149,7 +149,7 @@ public CommonResponse saveOrUpdate(BaseRepository "Seatunnel"，LocalExecutor -> "Local"。 + * + * Common base for all executor plugins. + * - Type is fixed to EXECUTOR. + * - Name auto-derives from the subclass simple name by stripping the trailing "Executor". + * e.g. SeatunnelExecutor -> "Seatunnel", LocalExecutor -> "Local". + */ +abstract class ExecutorPlugin : Plugin() +{ + override fun getType(): PluginType = PluginType.EXECUTOR + + override fun getName(): String = this::class.java.simpleName.removeSuffix("Executor") +} diff --git a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/ExecutorService.kt b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/ExecutorService.kt index 298992e96e..da17898fe9 100644 --- a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/ExecutorService.kt +++ b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/ExecutorService.kt @@ -9,15 +9,4 @@ interface ExecutorService : Service fun start(request: ExecutorRequest): ExecutorResponse fun stop(request: ExecutorRequest): ExecutorResponse - - /** - * 当前 Executor 的短名。默认实现：去掉类名末尾的 "ExecutorService"。 - * 例如 LocalExecutorService -> "Local"，SeatunnelExecutorService -> "Seatunnel"。 - * 大小写转换由调用方按业务需要处理。 - * - * Short name of this executor. Default: strip "ExecutorService" suffix from the class name. - * e.g. LocalExecutorService -> "Local", SeatunnelExecutorService -> "Seatunnel". - * Case conversion is the caller's responsibility. - */ - fun name(): String = this::class.java.simpleName.removeSuffix("ExecutorService") } diff --git a/pom.xml b/pom.xml index 5f3937a689..c9c2c14328 100644 --- a/pom.xml +++ b/pom.xml @@ -473,6 +473,9 @@ ${kotlin.version} 11 + + -Xjvm-default=all + diff --git a/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/local/LocalExecutorPluginTest.kt b/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/local/LocalExecutorPluginTest.kt index f1b50b6b44..81e9c87ef0 100644 --- a/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/local/LocalExecutorPluginTest.kt +++ b/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/local/LocalExecutorPluginTest.kt @@ -9,7 +9,7 @@ import org.junit.Test class LocalExecutorPluginTest { private val pluginManager: PluginManager - private val pluginName = "LocalExecutor" + private val pluginName = "Local" init { diff --git a/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/local/LocalExecutorServiceTest.kt b/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/local/LocalExecutorServiceTest.kt index 3fe0f2f155..1d3b30075b 100644 --- a/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/local/LocalExecutorServiceTest.kt +++ b/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/local/LocalExecutorServiceTest.kt @@ -18,7 +18,7 @@ import java.util.* class LocalExecutorServiceTest { private val pluginManager: PluginManager - private val pluginName = "LocalExecutor" + private val pluginName = "Local" init { diff --git a/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/seatunnel/SeatunnelExecutorPluginTest.kt b/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/seatunnel/SeatunnelExecutorPluginTest.kt index d7a47c8695..28aec3f097 100644 --- a/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/seatunnel/SeatunnelExecutorPluginTest.kt +++ b/test/datacap-test-executor/src/test/kotlin/io/edurt/datacap/test/seatunnel/SeatunnelExecutorPluginTest.kt @@ -9,7 +9,7 @@ import org.junit.Test class SeatunnelExecutorPluginTest { private val pluginManager: PluginManager - private val pluginName = "SeatunnelExecutor" + private val pluginName = "Seatunnel" init { From b2699e8949355977b74f9d1f87bc0f1d21081246 Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Thu, 21 May 2026 11:38:14 +0800 Subject: [PATCH 07/23] =?UTF-8?q?feat(dataset):=20=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E5=A2=9E=E5=8A=A0=E6=80=BB=E6=95=B0=20/=20?= =?UTF-8?q?=E5=B7=B2=E5=AE=8C=E6=88=90=20/=20=E8=BF=9B=E5=BA=A6=EF=BC=8C?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E6=9C=9F=E9=97=B4=E5=AE=9E=E6=97=B6=E5=88=B7?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E9=9B=86=20totalRows=20/=20totalSize?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - schema 2026.0.0/schema.sql: datacap_dataset_history 新增 total_count / processed_count / progress - DatasetHistoryEntity 同步加 totalCount / processedCount / progress 字段 - executor-spi 新增 ExecutorProgressListener，ExecutorRequest 加 preCount + progressListener - LocalExecutorService 支持 pre-count + 每 1000 行回调进度（PROGRESS_INTERVAL 从 10000 调小） - DataSetServiceImpl 进度回调实时更新 history 与 dataset.totalRows；totalSize 走独立 daemon 线程异步刷 ClickHouse system.parts，避免阻塞 MySQL 流式 cursor 触发 net_write_timeout - datacap.executor.local.preCount 默认 false：SELECT COUNT(*) FROM (userQuery) 在 MySQL/InnoDB 上会物化整个 derived table，慢于真正的同步 - DatasetHistory.vue 增加进度条 slot，DatasetUtils.ts 加 3 列，i18n 加 dataset.history.totalCount / processedCount / progress 三个 key --- configure/docker/application.properties | 6 ++ configure/etc/conf/application.properties | 6 ++ .../etc/conf/i18n/messages_en.properties | 3 + .../etc/conf/i18n/messages_zh-cn.properties | 3 + configure/schema/2026.0.0/schema.sql | 7 ++ .../service/entity/DatasetHistoryEntity.java | 11 +++ .../service/impl/DataSetServiceImpl.java | 96 ++++++++++++++++++- .../pages/admin/dataset/DatasetHistory.vue | 24 +++++ .../views/pages/admin/dataset/DatasetUtils.ts | 3 + .../executor/local/LocalExecutorService.kt | 81 ++++++++++++++-- .../configure/ExecutorProgressListener.kt | 13 +++ .../executor/configure/ExecutorRequest.kt | 4 +- 12 files changed, 244 insertions(+), 13 deletions(-) create mode 100644 executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorProgressListener.kt diff --git a/configure/docker/application.properties b/configure/docker/application.properties index 8985e7c9bc..65d724bf8c 100644 --- a/configure/docker/application.properties +++ b/configure/docker/application.properties @@ -53,6 +53,12 @@ datacap.executor.seatunnel.home=/opt/lib/seatunnel datacap.executor.local.fetchSize=1000 # Rows per batch flushed to the target via PreparedStatement.executeBatch(). datacap.executor.local.batchSize=1000 +# Whether to run SELECT COUNT(*) over the user query before sync to populate the progress denominator. +# Default OFF: wrapping the user query in SELECT COUNT(*) FROM (...) forces a derived-table +# materialization on MySQL/InnoDB which is slower than the sync itself for large tables. +# When OFF, progress bar shows the processed row count only (no percentage until done). +# Turn ON only when the source query is cheap to count (small / well-indexed / WHERE filtered). +datacap.executor.local.preCount=false # ------ Apache Seatunnel for Flink ------ # # datacap.executor.data= diff --git a/configure/etc/conf/application.properties b/configure/etc/conf/application.properties index a31d3d1925..3eee2becbf 100644 --- a/configure/etc/conf/application.properties +++ b/configure/etc/conf/application.properties @@ -53,6 +53,12 @@ datacap.executor.seatunnel.home=/opt/lib/seatunnel datacap.executor.local.fetchSize=1000 # Rows per batch flushed to the target via PreparedStatement.executeBatch(). datacap.executor.local.batchSize=1000 +# Whether to run SELECT COUNT(*) over the user query before sync to populate the progress denominator. +# Default OFF: wrapping the user query in SELECT COUNT(*) FROM (...) forces a derived-table +# materialization on MySQL/InnoDB which is slower than the sync itself for large tables. +# When OFF, progress bar shows the processed row count only (no percentage until done). +# Turn ON only when the source query is cheap to count (small / well-indexed / WHERE filtered). +datacap.executor.local.preCount=false # ------ Apache Seatunnel for Flink ------ # # datacap.executor.data= diff --git a/configure/etc/conf/i18n/messages_en.properties b/configure/etc/conf/i18n/messages_en.properties index e3f90d7e72..56e078540c 100644 --- a/configure/etc/conf/i18n/messages_en.properties +++ b/configure/etc/conf/i18n/messages_en.properties @@ -670,6 +670,9 @@ dataset.common.lifeCycleDay=Day dataset.common.lifeCycleHour=Hour dataset.common.notSpecifiedTitle=Not Specified dataset.common.history=Sync History +dataset.history.totalCount=Total +dataset.history.processedCount=Processed +dataset.history.progress=Progress dataset.common.clearData=Clear Data dataset.common.error=View Errors dataset.common.info=View Details diff --git a/configure/etc/conf/i18n/messages_zh-cn.properties b/configure/etc/conf/i18n/messages_zh-cn.properties index cc18988816..36b4cd782a 100644 --- a/configure/etc/conf/i18n/messages_zh-cn.properties +++ b/configure/etc/conf/i18n/messages_zh-cn.properties @@ -670,6 +670,9 @@ dataset.common.lifeCycleDay=\u5929 dataset.common.lifeCycleHour=\u5C0F\u65F6 dataset.common.notSpecifiedTitle=\u672A\u6307\u5B9A dataset.common.history=\u540C\u6B65\u5386\u53F2 +dataset.history.totalCount=\u603B\u6570 +dataset.history.processedCount=\u5DF2\u5B8C\u6210 +dataset.history.progress=\u8FDB\u5EA6 dataset.common.clearData=\u6E05\u9664\u6570\u636E dataset.common.error=\u67E5\u770B\u9519\u8BEF dataset.common.info=\u67E5\u770B\u8BE6\u60C5 diff --git a/configure/schema/2026.0.0/schema.sql b/configure/schema/2026.0.0/schema.sql index 14d6cf7de2..454bb1a0af 100644 --- a/configure/schema/2026.0.0/schema.sql +++ b/configure/schema/2026.0.0/schema.sql @@ -23,3 +23,10 @@ WHERE `executor` = 'LocalExecutor'; UPDATE `datacap_workflow` SET `executor` = 'Seatunnel' WHERE `executor` = 'SeatunnelExecutor'; + +-- 同步历史新增 总数 / 已完成 / 进度 三列 +-- Sync history adds total count / processed count / progress columns +ALTER TABLE `datacap_dataset_history` + ADD COLUMN `total_count` BIGINT DEFAULT NULL COMMENT 'Total rows from source query, NULL if pre-count is disabled', + ADD COLUMN `processed_count` BIGINT DEFAULT NULL COMMENT 'Rows successfully written to target', + ADD COLUMN `progress` DECIMAL(5, 2) DEFAULT NULL COMMENT 'processed_count / total_count * 100, NULL when total unknown'; diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/entity/DatasetHistoryEntity.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/entity/DatasetHistoryEntity.java index 9c28afb249..e5bff9b78b 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/entity/DatasetHistoryEntity.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/entity/DatasetHistoryEntity.java @@ -20,6 +20,8 @@ import javax.persistence.ManyToOne; import javax.persistence.Table; +import java.math.BigDecimal; + @Data @SuperBuilder @NoArgsConstructor @@ -44,6 +46,15 @@ public class DatasetHistoryEntity @Column(name = "count") private int count; + @Column(name = "total_count") + private Long totalCount; + + @Column(name = "processed_count") + private Long processedCount; + + @Column(name = "progress") + private BigDecimal progress; + @Column(name = "query_mode") @Enumerated(EnumType.STRING) private QueryMode mode; diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java index a47c82e66e..57f325a645 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java @@ -102,6 +102,9 @@ public class DataSetServiceImpl implements DataSetService { + // 同步期间每 SIZE_REFRESH_EVERY 次进度回调才刷一次 totalSize（查 ClickHouse system.parts 较重） + private static final int SIZE_REFRESH_EVERY = 10; + public final DataSetColumnRepository columnRepository; private final DataSetRepository repository; private final DatasetHistoryRepository historyRepository; @@ -1047,6 +1050,7 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut int fetchSize = Integer.parseInt(environment.getProperty("datacap.executor.local.fetchSize", "1000")); int batchSize = Integer.parseInt(environment.getProperty("datacap.executor.local.batchSize", "1000")); + boolean preCount = Boolean.parseBoolean(environment.getProperty("datacap.executor.local.preCount", "true")); ExecutorRequest request = new ExecutorRequest( taskName, entity.getUser().getUsername(), @@ -1062,18 +1066,104 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut RunEngine.valueOf(requireNonNull(environment.getProperty("datacap.executor.engine"))), null, fetchSize, - batchSize + batchSize, + preCount, + null ); + // 进度回调：每个 batch 写完后更新 datacap_dataset_history 以及数据集自身的 totalRows / totalSize + // totalRows 直接用已写入计数；totalSize 必须查 ClickHouse system.parts，开销大且不可控， + // 必须放到独立线程异步执行，否则会阻塞 MySQL 流式 cursor 触发 net_write_timeout + final DatasetHistoryEntity progressHistory = history; + final DataSetEntity progressEntity = entity; + final java.util.concurrent.atomic.AtomicInteger progressTick = new java.util.concurrent.atomic.AtomicInteger(); + final java.util.concurrent.atomic.AtomicBoolean sizeRefreshInFlight = new java.util.concurrent.atomic.AtomicBoolean(false); + final java.util.concurrent.ExecutorService sizeRefreshExecutor = Executors.newSingleThreadExecutor(r -> { + Thread t = new Thread(r, "dataset-size-refresh-" + taskName); + t.setDaemon(true); + return t; + }); + request.setProgressListener((processed, total) -> { + progressHistory.setProcessedCount(processed); + if (total >= 0) { + progressHistory.setTotalCount(total); + if (total > 0) { + java.math.BigDecimal percent = java.math.BigDecimal + .valueOf(processed * 100.0 / total) + .setScale(2, java.math.RoundingMode.HALF_UP); + progressHistory.setProgress(percent); + } + } + try { + historyRepository.save(progressHistory); + } + catch (Exception ex) { + log.warn("Update sync history progress failed: {}", ex.getMessage()); + } + // 同步期间用已写入行数滚动刷新 dataset.totalRows，方便列表实时看到进度 + try { + progressEntity.setTotalRows(String.valueOf(processed)); + repository.save(progressEntity); + } + catch (Exception ex) { + log.warn("Update dataset totalRows failed: {}", ex.getMessage()); + } + // 节流 + 异步刷新 totalSize：每 SIZE_REFRESH_EVERY 次 progress 提交一次到独立线程； + // 上一次还没回来就跳过，绝不阻塞主流（MySQL 真流式 cursor 对客户端读速度敏感） + if (progressTick.incrementAndGet() % SIZE_REFRESH_EVERY == 0 + && sizeRefreshInFlight.compareAndSet(false, true)) { + sizeRefreshExecutor.submit(() -> { + try { + this.flushTableMetadata( + progressEntity, + outputPlugin, + database, + requireNonNull(output.getOriginConfigure()), + outPlugin + ); + } + catch (Exception ex) { + log.warn("Refresh dataset totalSize failed: {}", ex.getMessage()); + } + finally { + sizeRefreshInFlight.set(false); + } + }); + } + }); + history.setState(RunState.RUNNING); historyRepository.save(history); - ExecutorResponse response = executorService.start(request); + ExecutorResponse response; + try { + response = executorService.start(request); + } + finally { + // 关闭异步 size 刷新线程，等正在跑的最多 30 秒 + sizeRefreshExecutor.shutdown(); + try { + if (!sizeRefreshExecutor.awaitTermination(30, java.util.concurrent.TimeUnit.SECONDS)) { + sizeRefreshExecutor.shutdownNow(); + } + } + catch (InterruptedException ie) { + sizeRefreshExecutor.shutdownNow(); + Thread.currentThread().interrupt(); + } + } history.setUpdateTime(new Date()); history.setElapsed((history.getUpdateTime().getTime() - history.getCreateTime().getTime()) / 1000); history.setMode(QueryMode.SYNC); history.setCount(response.getCount()); history.setState(response.getState()); + if (response.getSuccessful()) { + history.setProcessedCount((long) response.getCount()); + if (history.getTotalCount() == null || history.getTotalCount() < 0L) { + history.setTotalCount((long) response.getCount()); + } + history.setProgress(java.math.BigDecimal.valueOf(100.0).setScale(2, java.math.RoundingMode.HALF_UP)); + } Preconditions.checkArgument(response.getSuccessful(), response.getMessage()); this.flushTableMetadata( @@ -1086,7 +1176,7 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut }); }, () -> { - log.warn("Executor [ {} ] not found", entity.getExecutor()); + log.error("Executor [ {} ] not found", entity.getExecutor()); history.setMessage(String.format("Executor [ %s ] not found", entity.getExecutor())); history.setState(RunState.FAILURE); historyRepository.save(history); diff --git a/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue b/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue index d35ada2670..335d8ff032 100644 --- a/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue +++ b/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue @@ -24,6 +24,16 @@ {{ getStateText(row?.state) }} + + [ { key: 'id', label: t('common.id') }, { key: 'elapsed', label: t('common.elapsed') }, + { key: 'totalCount', label: t('dataset.history.totalCount') }, + { key: 'processedCount', label: t('dataset.history.processedCount') }, + { key: 'progress', label: t('dataset.history.progress'), slot: 'progress' }, { key: 'count', label: t('common.count') }, { key: 'createTime', label: t('common.createTime') }, { key: 'updateTime', label: t('common.updateTime') }, diff --git a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt index 42f3959ec1..83330cef10 100644 --- a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt +++ b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt @@ -9,6 +9,7 @@ import io.edurt.datacap.common.sql.configure.SqlColumn import io.edurt.datacap.common.sql.configure.SqlType import io.edurt.datacap.executor.ExecutorService import io.edurt.datacap.executor.common.RunState +import io.edurt.datacap.executor.configure.ExecutorProgressListener import io.edurt.datacap.executor.configure.ExecutorRequest import io.edurt.datacap.executor.configure.ExecutorResponse import io.edurt.datacap.executor.configure.OriginColumn @@ -56,21 +57,33 @@ class LocalExecutorService : ExecutorService val fetchSize = if (request.fetchSize > 0) request.fetchSize else DEFAULT_FETCH_SIZE val batchSize = if (request.batchSize > 0) request.batchSize else DEFAULT_BATCH_SIZE + val progressListener = request.progressListener taskLog.info( - "Resolved input plugin={} output plugin={} streaming={}/{} fetchSize={} batchSize={}", + "Resolved input plugin={} output plugin={} streaming={}/{} fetchSize={} batchSize={} preCount={}", inputPlugin.name(), outputPlugin.name(), inputPlugin.supportsStreaming(), outputPlugin.supportsStreaming(), - fetchSize, batchSize + fetchSize, batchSize, request.preCount ) taskLog.info("Source query: {}", query) + // 可选：先跑 SELECT COUNT(*) 拿到源端总行数，作为进度分母 + // Optional: pre-count to populate the total row count used for progress percentage + val totalCount: Long = if (request.preCount) preCount(inputPlugin, inputConfigure, query, taskLog) else -1L + if (totalCount >= 0) + { + taskLog.info("Pre-count: source total rows = {}", totalCount) + // 报告 0 进度，让上层立即知道 total + progressListener?.onProgress(0L, totalCount) + } + val written = if (inputPlugin.supportsStreaming() && outputPlugin.supportsStreaming()) { runStreaming( inputPlugin, inputConfigure, query, fetchSize, outputPlugin, outputConfigure, database, table, - targetColumns, sourceKeys, batchSize, taskLog + targetColumns, sourceKeys, batchSize, taskLog, + totalCount, progressListener ) } else @@ -78,13 +91,16 @@ class LocalExecutorService : ExecutorService runLegacy( inputPlugin, inputConfigure, query, outputPlugin, outputConfigure, database, table, - originColumns, batchSize, taskLog + originColumns, batchSize, taskLog, + totalCount, progressListener ) } response.count = if (written > Int.MAX_VALUE.toLong()) Int.MAX_VALUE else written.toInt() response.successful = true response.state = RunState.SUCCESS + // 最终上报一次：确保 100% + progressListener?.onProgress(written, if (totalCount >= 0) totalCount else written) taskLog.info("Local executor task completed: rows={} state=SUCCESS", written) } catch (ex: Exception) @@ -130,6 +146,44 @@ class LocalExecutorService : ExecutorService return ExecutorResponse(false, true, RunState.SUCCESS, null) } + /** + * 跑 SELECT COUNT(*) FROM (userQuery) 估算总行数。失败时返回 -1，不影响后续流程。 + * Run SELECT COUNT(*) over the user's query. On failure returns -1 — progress just lacks the denominator. + */ + private fun preCount(plugin: PluginService, configure: Configure, query: String, taskLog: Logger): Long + { + return try + { + val trimmed = query.trim().trimEnd(';') + val sql = "SELECT COUNT(*) AS total FROM ($trimmed) datacap_precount_t" + taskLog.info("Pre-count starting: {}", sql) + val t0 = System.nanoTime() + val result = plugin.execute(configure, sql) + val elapsedMs = (System.nanoTime() - t0) / 1_000_000 + taskLog.info("Pre-count returned in {} ms", elapsedMs) + if (result.isSuccessful != true) + { + taskLog.warn("Pre-count failed, will run without total: {}", result.message) + return -1L + } + val rows = result.columns ?: return -1L + if (rows.isEmpty()) return -1L + val first = rows[0] + when (first) + { + is com.fasterxml.jackson.databind.node.ObjectNode -> first.get("total")?.asLong(-1L) ?: -1L + is List<*> -> (first.firstOrNull() as? Number)?.toLong() ?: -1L + is Number -> first.toLong() + else -> -1L + } + } + catch (ex: Exception) + { + taskLog.warn("Pre-count threw, will run without total: {}", ex.message) + -1L + } + } + /** * 流式路径：源端 fetchSize 拉取，目标端 PreparedStatement 批量写。 * 全程不在 JVM 内物化整个结果集。 @@ -146,12 +200,14 @@ class LocalExecutorService : ExecutorService targetColumns: List, sourceKeys: List, batchSize: Int, - taskLog: Logger + taskLog: Logger, + totalCount: Long, + progressListener: ExecutorProgressListener? ): Long { taskLog.info( - "Streaming sync start: target=`{}`.`{}` columns={} fetchSize={} batchSize={}", - database, table, targetColumns.size, fetchSize, batchSize + "Streaming sync start: target=`{}`.`{}` columns={} fetchSize={} batchSize={} total={}", + database, table, targetColumns.size, fetchSize, batchSize, totalCount ) val startNanos = System.nanoTime() var written = 0L @@ -188,6 +244,7 @@ class LocalExecutorService : ExecutorService "Streaming sync progress: read={} committed={} elapsed={}s rps={}", written, writer.writtenCount(), "%.1f".format(seconds), rps ) + progressListener?.onProgress(writer.writtenCount(), totalCount) } } }) @@ -214,7 +271,9 @@ class LocalExecutorService : ExecutorService table: String, originColumns: List, batchSize: Int, - taskLog: Logger + taskLog: Logger, + totalCount: Long, + progressListener: ExecutorProgressListener? ): Long { taskLog.info("Legacy sync start: target=`{}`.`{}` batchSize={}", database, table, batchSize) @@ -226,6 +285,8 @@ class LocalExecutorService : ExecutorService } val rows = inputResult.columns ?: return 0L taskLog.info("Legacy sync: source materialized {} rows", rows.size) + // 全量路径已经能拿到精确总数，覆盖 pre-count 的估算 + val effectiveTotal = if (totalCount >= 0) totalCount else rows.size.toLong() // 目标端能流式：用 BatchWriter 安全 + 节省内存 if (outputPlugin.supportsStreaming()) @@ -250,6 +311,7 @@ class LocalExecutorService : ExecutorService if (written % PROGRESS_INTERVAL == 0L) { taskLog.info("Legacy sync progress: written={} committed={}", written, writer.writtenCount()) + progressListener?.onProgress(writer.writtenCount(), effectiveTotal) } } } @@ -289,6 +351,7 @@ class LocalExecutorService : ExecutorService if (written % PROGRESS_INTERVAL == 0L) { taskLog.info("Legacy sync progress (sql batch): written={}", written) + progressListener?.onProgress(written, effectiveTotal) } } } @@ -384,6 +447,6 @@ class LocalExecutorService : ExecutorService { private const val DEFAULT_FETCH_SIZE = 1000 private const val DEFAULT_BATCH_SIZE = 1000 - private const val PROGRESS_INTERVAL = 10_000L + private const val PROGRESS_INTERVAL = 1_000L } } diff --git a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorProgressListener.kt b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorProgressListener.kt new file mode 100644 index 0000000000..b5b80a162b --- /dev/null +++ b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorProgressListener.kt @@ -0,0 +1,13 @@ +package io.edurt.datacap.executor.configure + +/** + * 执行器进度上报回调。执行器按一定行数节奏调用。 + * Progress listener called by the executor at a fixed row-count cadence. + * + * @param processed 已成功写入目标的行数 / rows already written to the target + * @param total 源端总行数；未知时传 -1 / source total rows; pass -1 when unknown + */ +fun interface ExecutorProgressListener +{ + fun onProgress(processed: Long, total: Long) +} diff --git a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorRequest.kt b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorRequest.kt index b88b4408a9..7af6e66af9 100644 --- a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorRequest.kt +++ b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/configure/ExecutorRequest.kt @@ -22,7 +22,9 @@ data class ExecutorRequest @JvmOverloads constructor( var runEngine: RunEngine = RunEngine.SPARK, var transform: ExecutorConfigure? = null, var fetchSize: Int = 1000, - var batchSize: Int = 1000 + var batchSize: Int = 1000, + var preCount: Boolean = false, + var progressListener: ExecutorProgressListener? = null ) { constructor( From 4442ac88a6771b86eb534f4a53b0fb0f38d2293c Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Thu, 21 May 2026 13:05:20 +0800 Subject: [PATCH 08/23] =?UTF-8?q?feat(dataset):=20=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E6=94=AF=E6=8C=81=E6=9F=A5=E7=9C=8B=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E6=97=A5=E5=BF=97=EF=BC=88=E5=AE=9E=E6=97=B6=20+=20?= =?UTF-8?q?=E5=B7=B2=E7=BB=93=E6=9D=9F=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - schema 2026.0.0/schema.sql: datacap_dataset_history 新增 task_name / work_home 两列，用于定位 executor 写出的独立日志文件 - DatasetHistoryEntity 同步加 taskName / workHome 字段 - DataSetServiceImpl 创建 history 时把 taskName / workHome 落库；新增 getHistoryLog(id) 读取 {workHome}/{taskName}.log，含路径穿越防御 - DataSetController 加 GET /api/v1/dataset/history/log/{id} - 新增 DatasetHistoryLogger.vue，复用 SeaTunnel 用的 ShadcnLogger；RUNNING 任务默认开启 3 秒自动刷新 - DatasetHistory.vue 加查看日志按钮列；DatasetUtils.ts historyHeaders 加 action - i18n: dataset.history.viewLog / logger / autoRefresh / refreshInterval（en + zh-cn） --- .../etc/conf/i18n/messages_en.properties | 4 + .../etc/conf/i18n/messages_zh-cn.properties | 4 + configure/schema/2026.0.0/schema.sql | 6 + .../server/controller/DataSetController.java | 6 + .../service/entity/DatasetHistoryEntity.java | 6 + .../service/service/DataSetService.java | 6 + .../service/impl/DataSetServiceImpl.java | 42 +++++ core/datacap-ui/src/services/dataset.ts | 11 ++ .../pages/admin/dataset/DatasetHistory.vue | 27 ++- .../admin/dataset/DatasetHistoryLogger.vue | 155 ++++++++++++++++++ .../views/pages/admin/dataset/DatasetUtils.ts | 3 +- 11 files changed, 268 insertions(+), 2 deletions(-) create mode 100644 core/datacap-ui/src/views/pages/admin/dataset/DatasetHistoryLogger.vue diff --git a/configure/etc/conf/i18n/messages_en.properties b/configure/etc/conf/i18n/messages_en.properties index 56e078540c..a90a693364 100644 --- a/configure/etc/conf/i18n/messages_en.properties +++ b/configure/etc/conf/i18n/messages_en.properties @@ -673,6 +673,10 @@ dataset.common.history=Sync History dataset.history.totalCount=Total dataset.history.processedCount=Processed dataset.history.progress=Progress +dataset.history.viewLog=View Log +dataset.history.logger=Sync Log +dataset.history.autoRefresh=Auto Refresh +dataset.history.refreshInterval=refresh every {seconds}s dataset.common.clearData=Clear Data dataset.common.error=View Errors dataset.common.info=View Details diff --git a/configure/etc/conf/i18n/messages_zh-cn.properties b/configure/etc/conf/i18n/messages_zh-cn.properties index 36b4cd782a..a594b80f35 100644 --- a/configure/etc/conf/i18n/messages_zh-cn.properties +++ b/configure/etc/conf/i18n/messages_zh-cn.properties @@ -673,6 +673,10 @@ dataset.common.history=\u540C\u6B65\u5386\u53F2 dataset.history.totalCount=\u603B\u6570 dataset.history.processedCount=\u5DF2\u5B8C\u6210 dataset.history.progress=\u8FDB\u5EA6 +dataset.history.viewLog=\u67E5\u770B\u65E5\u5FD7 +dataset.history.logger=\u540C\u6B65\u65E5\u5FD7 +dataset.history.autoRefresh=\u81EA\u52A8\u5237\u65B0 +dataset.history.refreshInterval=\u6BCF {seconds} \u79D2\u5237\u65B0 dataset.common.clearData=\u6E05\u9664\u6570\u636E dataset.common.error=\u67E5\u770B\u9519\u8BEF dataset.common.info=\u67E5\u770B\u8BE6\u60C5 diff --git a/configure/schema/2026.0.0/schema.sql b/configure/schema/2026.0.0/schema.sql index 454bb1a0af..43572ddaf7 100644 --- a/configure/schema/2026.0.0/schema.sql +++ b/configure/schema/2026.0.0/schema.sql @@ -30,3 +30,9 @@ ALTER TABLE `datacap_dataset_history` ADD COLUMN `total_count` BIGINT DEFAULT NULL COMMENT 'Total rows from source query, NULL if pre-count is disabled', ADD COLUMN `processed_count` BIGINT DEFAULT NULL COMMENT 'Rows successfully written to target', ADD COLUMN `progress` DECIMAL(5, 2) DEFAULT NULL COMMENT 'processed_count / total_count * 100, NULL when total unknown'; + +-- 同步历史新增 任务名 / 工作目录，用于定位 executor 写出的独立任务日志文件 +-- Sync history adds task name / work home so the UI can locate the executor's task log file +ALTER TABLE `datacap_dataset_history` + ADD COLUMN `task_name` VARCHAR(64) DEFAULT NULL COMMENT 'Executor task name; also the log file basename', + ADD COLUMN `work_home` VARCHAR(512) DEFAULT NULL COMMENT 'Executor task workHome; logs live at {work_home}/{task_name}.log'; diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/DataSetController.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/DataSetController.java index 42a2a60085..3baecd85ba 100644 --- a/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/DataSetController.java +++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/DataSetController.java @@ -75,6 +75,12 @@ public CommonResponse> history(@PathVariable St return this.service.getHistory(code, filter); } + @GetMapping(value = "history/log/{id}") + public CommonResponse> historyLog(@PathVariable Long id) + { + return this.service.getHistoryLog(id); + } + @GetMapping(value = "getActuators") public CommonResponse> getActuators() { diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/entity/DatasetHistoryEntity.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/entity/DatasetHistoryEntity.java index e5bff9b78b..aec2aaa81f 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/entity/DatasetHistoryEntity.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/entity/DatasetHistoryEntity.java @@ -55,6 +55,12 @@ public class DatasetHistoryEntity @Column(name = "progress") private BigDecimal progress; + @Column(name = "task_name") + private String taskName; + + @Column(name = "work_home") + private String workHome; + @Column(name = "query_mode") @Enumerated(EnumType.STRING) private QueryMode mode; diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/DataSetService.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/DataSetService.java index dd1917f880..9126e35c38 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/DataSetService.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/DataSetService.java @@ -31,4 +31,10 @@ public interface DataSetService CommonResponse getInfo(String code); CommonResponse> getHistory(String code, FilterBody filter); + + /** + * 读取指定同步历史的运行日志（独立任务日志文件，由 executor 写入 workHome） + * Read the executor's task log for the given sync history record + */ + CommonResponse> getHistoryLog(Long id); } diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java index 57f325a645..ad59c993a7 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java @@ -324,6 +324,44 @@ public CommonResponse> getHistory(String code, .orElse(CommonResponse.failure(String.format("DataSet [ %s ] not found", code))); } + @Override + public CommonResponse> getHistoryLog(Long id) + { + return historyRepository.findById(id) + .map(history -> { + String workHome = history.getWorkHome(); + String taskName = history.getTaskName(); + if (StringUtils.isBlank(workHome) || StringUtils.isBlank(taskName)) { + return CommonResponse.>failure("Sync history has no log location (older record or non-local executor)"); + } + // 防穿越：拼好后规整为绝对路径，并校验仍在 dataHome 下 + java.io.File logFile = new java.io.File(workHome, taskName + ".log"); + String dataHome = initializerConfigure.getDataHome(); + try { + String canonicalLog = logFile.getCanonicalPath(); + String canonicalHome = new java.io.File(dataHome).getCanonicalPath(); + if (!canonicalLog.startsWith(canonicalHome)) { + return CommonResponse.>failure("Illegal log path"); + } + } + catch (java.io.IOException ex) { + return CommonResponse.>failure("Resolve log path failed: " + ex.getMessage()); + } + if (!logFile.exists()) { + return CommonResponse.>success(Lists.newArrayList()); + } + try (java.io.FileInputStream stream = new java.io.FileInputStream(logFile)) { + List lines = org.apache.commons.io.IOUtils.readLines(stream, java.nio.charset.StandardCharsets.UTF_8); + return CommonResponse.>success(lines); + } + catch (java.io.IOException ex) { + log.error("Read sync log [ {} ] failed", logFile.getAbsolutePath(), ex); + return CommonResponse.>failure("Read log file failed: " + ex.getMessage()); + } + }) + .orElseGet(() -> CommonResponse.>failure(String.format("Sync history [ %d ] not found", id))); + } + @Override @Transactional @SendNotification(type = NotificationType.DELETED) @@ -1047,6 +1085,10 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut entity.getUser().getUsername(), String.join(File.separator, "dataset", "executor", executorKey, taskName) ); + // 把 taskName / workHome 落到 history，供 UI 定位日志文件 + history.setTaskName(taskName); + history.setWorkHome(workHome); + historyRepository.save(history); int fetchSize = Integer.parseInt(environment.getProperty("datacap.executor.local.fetchSize", "1000")); int batchSize = Integer.parseInt(environment.getProperty("datacap.executor.local.batchSize", "1000")); diff --git a/core/datacap-ui/src/services/dataset.ts b/core/datacap-ui/src/services/dataset.ts index e06eece3f6..9795951b2d 100644 --- a/core/datacap-ui/src/services/dataset.ts +++ b/core/datacap-ui/src/services/dataset.ts @@ -59,6 +59,17 @@ export class DatasetService return new HttpUtils().post(`${ DEFAULT_PATH }/history/${ code }`, configure) } + /** + * Read the executor's task log of a given sync history record. + * + * @param {number} id - the dataset history id + * @return {Promise} response containing the log lines (List) + */ + getHistoryLog(id: number): Promise + { + return new HttpUtils().get(`${ DEFAULT_PATH }/history/log/${ id }`) + } + /** * Sync data with the server using the provided id. * diff --git a/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue b/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue index 335d8ff032..a7645c07a5 100644 --- a/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue +++ b/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue @@ -34,8 +34,19 @@ {{ formatProgressText(row?.progress) }} + + + + + + + +
+ + {{ $t('dataset.history.autoRefresh') }} + + ({{ $t('dataset.history.refreshInterval', { seconds: refreshSeconds }) }}) + +
+ + + + + + + + diff --git a/core/datacap-ui/src/views/pages/admin/dataset/DatasetUtils.ts b/core/datacap-ui/src/views/pages/admin/dataset/DatasetUtils.ts index a2c68c8cf8..1a31dd8dd9 100644 --- a/core/datacap-ui/src/views/pages/admin/dataset/DatasetUtils.ts +++ b/core/datacap-ui/src/views/pages/admin/dataset/DatasetUtils.ts @@ -33,7 +33,8 @@ export function useDatasetHeaders() { key: 'count', label: t('common.count') }, { key: 'createTime', label: t('common.createTime') }, { key: 'updateTime', label: t('common.updateTime') }, - { key: 'state', label: t('common.state'), slot: 'state' } + { key: 'state', label: t('common.state'), slot: 'state' }, + { key: 'action', label: t('common.action'), slot: 'action' } ]) return { From d35d850166f6019978769f81d03608465f63e293 Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Thu, 21 May 2026 16:06:07 +0800 Subject: [PATCH 09/23] =?UTF-8?q?feat(dataset):=20=E6=94=AF=E6=8C=81?= =?UTF-8?q?=E6=89=8B=E5=8A=A8=E5=81=9C=E6=AD=A2=E6=AD=A3=E5=9C=A8=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E7=9A=84=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RunState 新增 STOPPING 中间态：用户点击停止到任务真正退出之间显示"停止中" - RowCallback 新增 onStatement(Statement) 回调，JdbcStreamAdapter 暴露 JDBC Statement - LocalExecutorService: - 静态 runningTasks: ConcurrentHashMap 索引活跃任务 - stop() 设置 cancelled 标志 + 异步调用 Statement.cancel() 立即 abort 源端流式 cursor - 行循环每行检查 cancelled；post-loop guard 兜底处理驱动 cancel 后 rs.next()=false 的情况 - 停止事件写入任务专属日志（Stop requested by user / Source JDBC statement cancelled） - catch 区分 TaskCancelledException 与 JDBC cancel 引发的 SQLException，均归入 STOPPED - DataSetService.stopHistory(id) + Controller PUT /api/v1/dataset/history/stop/{id} - stopHistory 在 executor.stop 成功后立即把 history.state 写成 STOPPING；进度回调期间维持 STOPPING 不会被覆盖回 RUNNING - 同步主流程 finally 块从 stoppingHistoryIds 移除；遇到 STOPPED 状态跳过 Preconditions / flushTableMetadata，保留已写入数据 - UI: 停止按钮仅在 RUNNING/CREATED 显示；STOPPING 状态加橙色 + 国际化（state.common.stopping / dataset.history.stop / stopRequested） --- .../etc/conf/i18n/messages_en.properties | 3 + .../etc/conf/i18n/messages_zh-cn.properties | 3 + .../server/controller/DataSetController.java | 6 + .../service/service/DataSetService.java | 6 + .../service/impl/DataSetServiceImpl.java | 60 ++++++++ .../spi/adapter/JdbcStreamAdapter.java | 2 + .../datacap/spi/adapter/RowCallback.java | 9 ++ core/datacap-ui/src/services/dataset.ts | 11 ++ core/datacap-ui/src/utils/common.ts | 4 + .../pages/admin/dataset/DatasetHistory.vue | 47 +++++- .../executor/local/LocalExecutorService.kt | 144 ++++++++++++++++-- .../edurt/datacap/executor/common/RunState.kt | 1 + 12 files changed, 282 insertions(+), 14 deletions(-) diff --git a/configure/etc/conf/i18n/messages_en.properties b/configure/etc/conf/i18n/messages_en.properties index a90a693364..6df5b948ca 100644 --- a/configure/etc/conf/i18n/messages_en.properties +++ b/configure/etc/conf/i18n/messages_en.properties @@ -13,6 +13,7 @@ state.common.running=Running state.common.success=Run Successful state.common.failure=Run Failed state.common.stop=Stopped +state.common.stopping=Stopping state.common.timeout=Run Timed Out state.common.queue=Queued ## Query i18n @@ -677,6 +678,8 @@ dataset.history.viewLog=View Log dataset.history.logger=Sync Log dataset.history.autoRefresh=Auto Refresh dataset.history.refreshInterval=refresh every {seconds}s +dataset.history.stop=Stop +dataset.history.stopRequested=Stop requested. The task will exit at the next row checkpoint. dataset.common.clearData=Clear Data dataset.common.error=View Errors dataset.common.info=View Details diff --git a/configure/etc/conf/i18n/messages_zh-cn.properties b/configure/etc/conf/i18n/messages_zh-cn.properties index a594b80f35..6059a8658b 100644 --- a/configure/etc/conf/i18n/messages_zh-cn.properties +++ b/configure/etc/conf/i18n/messages_zh-cn.properties @@ -13,6 +13,7 @@ state.common.running=\u8FD0\u884C\u4E2D state.common.success=\u8FD0\u884C\u6210\u529F state.common.failure=\u8FD0\u884C\u5931\u8D25 state.common.stop=\u5DF2\u505C\u6B62 +state.common.stopping=\u505C\u6B62\u4E2D state.common.timeout=\u8FD0\u884C\u8D85\u65F6 state.common.queue=\u6392\u961F\u4E2D ## Query i18n @@ -677,6 +678,8 @@ dataset.history.viewLog=\u67E5\u770B\u65E5\u5FD7 dataset.history.logger=\u540C\u6B65\u65E5\u5FD7 dataset.history.autoRefresh=\u81EA\u52A8\u5237\u65B0 dataset.history.refreshInterval=\u6BCF {seconds} \u79D2\u5237\u65B0 +dataset.history.stop=\u505C\u6B62 +dataset.history.stopRequested=\u5DF2\u8BF7\u6C42\u505C\u6B62\uFF0C\u4EFB\u52A1\u5C06\u5728\u4E0B\u4E00\u884C\u68C0\u67E5\u70B9\u9000\u51FA dataset.common.clearData=\u6E05\u9664\u6570\u636E dataset.common.error=\u67E5\u770B\u9519\u8BEF dataset.common.info=\u67E5\u770B\u8BE6\u60C5 diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/DataSetController.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/DataSetController.java index 3baecd85ba..6fe4b4e4c9 100644 --- a/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/DataSetController.java +++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/DataSetController.java @@ -81,6 +81,12 @@ public CommonResponse> historyLog(@PathVariable Long id) return this.service.getHistoryLog(id); } + @PutMapping(value = "history/stop/{id}") + public CommonResponse historyStop(@PathVariable Long id) + { + return this.service.stopHistory(id); + } + @GetMapping(value = "getActuators") public CommonResponse> getActuators() { diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/DataSetService.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/DataSetService.java index 9126e35c38..7e26f1c4d1 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/DataSetService.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/DataSetService.java @@ -37,4 +37,10 @@ public interface DataSetService * Read the executor's task log for the given sync history record */ CommonResponse> getHistoryLog(Long id); + + /** + * 停止正在运行的同步任务。仅对 state=RUNNING 且当前进程持有该 taskName 的任务生效 + * Stop a running sync task. Effective only when state=RUNNING and the task is held by this process + */ + CommonResponse stopHistory(Long id); } diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java index ad59c993a7..cf2c732ec7 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java @@ -105,6 +105,10 @@ public class DataSetServiceImpl // 同步期间每 SIZE_REFRESH_EVERY 次进度回调才刷一次 totalSize（查 ClickHouse system.parts 较重） private static final int SIZE_REFRESH_EVERY = 10; + // 已请求停止但还没真正停下来的 history id 集合：进度回调据此把 state 维持为 STOPPING 而不是 RUNNING + // Set of history ids whose sync has been asked to stop but hasn't reached its cancellation checkpoint yet. + private static final java.util.Set stoppingHistoryIds = java.util.concurrent.ConcurrentHashMap.newKeySet(); + public final DataSetColumnRepository columnRepository; private final DataSetRepository repository; private final DatasetHistoryRepository historyRepository; @@ -362,6 +366,48 @@ public CommonResponse> getHistoryLog(Long id) .orElseGet(() -> CommonResponse.>failure(String.format("Sync history [ %d ] not found", id))); } + @Override + public CommonResponse stopHistory(Long id) + { + log.info("Stop history [ {} ] requested", id); + return historyRepository.findById(id) + .map(history -> { + log.info("Stop history [ {} ] state={} taskName={}", id, history.getState(), history.getTaskName()); + if (history.getState() != RunState.RUNNING && history.getState() != RunState.CREATED) { + return CommonResponse.failure(String.format("Sync history [ %d ] is not running (current state: %s)", id, history.getState())); + } + String taskName = history.getTaskName(); + if (StringUtils.isBlank(taskName)) { + return CommonResponse.failure("Sync history has no taskName (older record), cannot stop"); + } + DataSetEntity dataset = history.getDataset(); + if (dataset == null || StringUtils.isBlank(dataset.getExecutor())) { + return CommonResponse.failure("Sync history has no associated dataset executor"); + } + log.info("Stop history [ {} ] resolving executor plugin [ {} ]", id, dataset.getExecutor()); + Optional executorPlugin = pluginManager.getPlugin(dataset.getExecutor()); + if (executorPlugin.isEmpty()) { + return CommonResponse.failure(String.format("Executor [ %s ] not found", dataset.getExecutor())); + } + ExecutorService executorService = executorPlugin.get().getService(ExecutorService.class); + // stop 只需要 taskName 定位活跃任务，input/output 用空 configure 占位 + ExecutorConfigure stopPlaceholder = new ExecutorConfigure(null); + ExecutorRequest stopRequest = new ExecutorRequest(taskName, "", stopPlaceholder, stopPlaceholder); + log.info("Stop history [ {} ] calling executor.stop(taskName={})", id, taskName); + ExecutorResponse stopResponse = executorService.stop(stopRequest); + log.info("Stop history [ {} ] executor returned successful={} message={}", id, stopResponse.getSuccessful(), stopResponse.getMessage()); + if (Boolean.TRUE.equals(stopResponse.getSuccessful())) { + // 让 UI 立刻看到 STOPPING 中间态；进度回调会继续把它保持住直到真正停下 + stoppingHistoryIds.add(id); + history.setState(RunState.STOPPING); + historyRepository.save(history); + return CommonResponse.success(Boolean.TRUE); + } + return CommonResponse.failure(stopResponse.getMessage() == null ? "Stop failed" : stopResponse.getMessage()); + }) + .orElseGet(() -> CommonResponse.failure(String.format("Sync history [ %d ] not found", id))); + } + @Override @Transactional @SendNotification(type = NotificationType.DELETED) @@ -1136,6 +1182,10 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut progressHistory.setProgress(percent); } } + // 收到停止请求后，进度回调不能把 state 写回 RUNNING；保持 STOPPING 直到真正退出 + if (stoppingHistoryIds.contains(progressHistory.getId())) { + progressHistory.setState(RunState.STOPPING); + } try { historyRepository.save(progressHistory); } @@ -1182,6 +1232,8 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut response = executorService.start(request); } finally { + // 任务一旦真正退出，从 STOPPING 集合移除（不管是正常 / 失败 / 被停止） + stoppingHistoryIds.remove(progressHistory.getId()); // 关闭异步 size 刷新线程，等正在跑的最多 30 秒 sizeRefreshExecutor.shutdown(); try { @@ -1199,6 +1251,9 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut history.setMode(QueryMode.SYNC); history.setCount(response.getCount()); history.setState(response.getState()); + if (response.getMessage() != null) { + history.setMessage(response.getMessage()); + } if (response.getSuccessful()) { history.setProcessedCount((long) response.getCount()); if (history.getTotalCount() == null || history.getTotalCount() < 0L) { @@ -1206,6 +1261,11 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut } history.setProgress(java.math.BigDecimal.valueOf(100.0).setScale(2, java.math.RoundingMode.HALF_UP)); } + historyRepository.save(history); + // STOPPED 是用户主动停止，已经把最终状态写入 history，无需当作异常抛出 / 也不刷 table metadata + if (response.getState() == RunState.STOPPED) { + return; + } Preconditions.checkArgument(response.getSuccessful(), response.getMessage()); this.flushTableMetadata( diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcStreamAdapter.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcStreamAdapter.java index 6fd8cabd07..52666fca9b 100644 --- a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcStreamAdapter.java +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/JdbcStreamAdapter.java @@ -59,6 +59,8 @@ public static void executeStream( ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) { applyFetchSize(statement, effectiveFetchSize, configure); + // 暴露给上层，方便取消逻辑直接调用 Statement.cancel() + callback.onStatement(statement); try (ResultSet rs = statement.executeQuery(sql)) { ResultSetMetaData metaData = rs.getMetaData(); int columnCount = metaData.getColumnCount(); diff --git a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/RowCallback.java b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/RowCallback.java index e8ac5aa26a..16c495b80e 100644 --- a/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/RowCallback.java +++ b/core/datacap-spi/src/main/java/io/edurt/datacap/spi/adapter/RowCallback.java @@ -1,5 +1,6 @@ package io.edurt.datacap.spi.adapter; +import java.sql.Statement; import java.util.List; public interface RowCallback @@ -10,6 +11,14 @@ public interface RowCallback */ default void onSchema(List headers, List types) {} + /** + * 在执行 query 之前回调一次，把底层 Statement 暴露给上层， + * 让取消逻辑可以直接调用 {@link Statement#cancel()} 中断阻塞中的 fetch。 + * Called once right before the underlying query executes, exposing the JDBC Statement so the caller + * can issue {@link Statement#cancel()} to interrupt a blocked fetch. + */ + default void onStatement(Statement statement) {} + /** * 每一行回调一次。 * Called per row. Values are aligned to {@code headers} from {@link #onSchema}. diff --git a/core/datacap-ui/src/services/dataset.ts b/core/datacap-ui/src/services/dataset.ts index 9795951b2d..d862021aad 100644 --- a/core/datacap-ui/src/services/dataset.ts +++ b/core/datacap-ui/src/services/dataset.ts @@ -70,6 +70,17 @@ export class DatasetService return new HttpUtils().get(`${ DEFAULT_PATH }/history/log/${ id }`) } + /** + * Stop a running sync task identified by the given history record id. + * + * @param {number} id - the dataset history id (must be RUNNING) + * @return {Promise} success when the cancel flag has been set + */ + stopHistory(id: number): Promise + { + return new HttpUtils().put(`${ DEFAULT_PATH }/history/stop/${ id }`) + } + /** * Sync data with the server using the provided id. * diff --git a/core/datacap-ui/src/utils/common.ts b/core/datacap-ui/src/utils/common.ts index 0bacd15da8..498eccc1a1 100644 --- a/core/datacap-ui/src/utils/common.ts +++ b/core/datacap-ui/src/utils/common.ts @@ -29,6 +29,8 @@ const getColor = (origin: string): string => { return 'hsl(142.1 76.2% 36.3%)' case 'FAILURE': return 'hsl(346.8 77.2% 49.8%)' + case 'STOPPING': + return 'hsl(38 92% 50%)' case 'STOPPED': return '#17233d' case 'TIMEOUT': @@ -79,6 +81,8 @@ export function useUtil() return t('state.common.success') case 'FAILURE': return t('state.common.failure') + case 'STOPPING': + return t('state.common.stopping') case 'STOPPED': return t('state.common.stop') case 'TIMEOUT': diff --git a/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue b/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue index a7645c07a5..106f993672 100644 --- a/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue +++ b/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue @@ -36,9 +36,18 @@ @@ -123,7 +132,8 @@ export default defineComponent({ pageSize: 10, dataCount: 0, loggerVisible: false, - loggerInfo: null as any + loggerInfo: null as any, + stoppingId: null as number | null } }, created() @@ -199,6 +209,35 @@ export default defineComponent({ { this.loggerVisible = false this.loggerInfo = null + }, + isStoppable(row: any): boolean + { + // STOPPING 状态已经在停了，不再展示按钮，防止重复点击 + return row?.state === 'RUNNING' || row?.state === 'CREATED' + }, + onStop(row: any) + { + if (!row?.id) { + return + } + this.stoppingId = row.id + DatasetService.stopHistory(row.id) + .then(response => { + if (response.status) { + this.$Message.success({ + content: this.$t('dataset.history.stopRequested'), + showIcon: true + }) + this.handleInitialize() + } + else { + this.$Message.error({ + content: response.message, + showIcon: true + }) + } + }) + .finally(() => this.stoppingId = null) } } }) diff --git a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt index 83330cef10..135bc5da7a 100644 --- a/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt +++ b/executor/datacap-executor-local/src/main/kotlin/io/edurt/datacap/executor/local/LocalExecutorService.kt @@ -21,17 +21,40 @@ import io.edurt.datacap.spi.adapter.RowCallback import io.edurt.datacap.spi.model.Configure import org.slf4j.Logger import org.slf4j.LoggerFactory +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicBoolean -@SuppressFBWarnings(value = ["BC_BAD_CAST_TO_ABSTRACT_COLLECTION", "RV_RETURN_VALUE_IGNORED_BAD_PRACTICE"]) +@SuppressFBWarnings(value = ["BC_BAD_CAST_TO_ABSTRACT_COLLECTION", "RV_RETURN_VALUE_IGNORED_BAD_PRACTICE", "MS_MUTABLE_COLLECTION_PKGPROTECT"]) class LocalExecutorService : ExecutorService { private val log = LoggerFactory.getLogger(LocalExecutorService::class.java) + /** + * 已注册的活跃任务句柄。 + * - cancelled: 调度循环每条行检查 + * - sourceStatement: stop() 会调 cancel() 立即终止源端 JDBC 查询，不必等下一行 + * - taskLog: 让 stop() 把"用户停止"事件写到任务专属日志里 + * - rowsAtStop: 记录被停止时已处理的行数，供 history 落库 + */ + private class TaskHandle + { + val cancelled: AtomicBoolean = AtomicBoolean(false) + @Volatile var sourceStatement: java.sql.Statement? = null + @Volatile var taskLog: Logger? = null + @Volatile var rowsAtStop: Long = 0L + } + override fun start(request: ExecutorRequest): ExecutorResponse { val response = ExecutorResponse() val loggerExecutor: LoggerExecutor<*>? = newTaskLogger(request) val taskLog: Logger = loggerExecutor?.getLogger() ?: log + val handle = TaskHandle() + handle.taskLog = taskLog + if (request.taskName.isNotBlank()) + { + runningTasks[request.taskName] = handle + } try { taskLog.info("Local executor task starting: task={} user={}", request.taskName, request.userName) @@ -83,7 +106,7 @@ class LocalExecutorService : ExecutorService inputPlugin, inputConfigure, query, fetchSize, outputPlugin, outputConfigure, database, table, targetColumns, sourceKeys, batchSize, taskLog, - totalCount, progressListener + totalCount, progressListener, handle ) } else @@ -92,7 +115,7 @@ class LocalExecutorService : ExecutorService inputPlugin, inputConfigure, query, outputPlugin, outputConfigure, database, table, originColumns, batchSize, taskLog, - totalCount, progressListener + totalCount, progressListener, handle ) } @@ -103,20 +126,48 @@ class LocalExecutorService : ExecutorService progressListener?.onProgress(written, if (totalCount >= 0) totalCount else written) taskLog.info("Local executor task completed: rows={} state=SUCCESS", written) } - catch (ex: Exception) + catch (ex: TaskCancelledException) { - taskLog.error("Local executor failed", ex) + taskLog.warn("Local executor task stopped by user: rows={} task={}", ex.processed, request.taskName) + response.count = if (ex.processed > Int.MAX_VALUE.toLong()) Int.MAX_VALUE else ex.processed.toInt() response.successful = false - response.state = RunState.FAILURE - response.message = ex.message + response.state = RunState.STOPPED + response.message = "Stopped by user" + } + catch (ex: Exception) + { + // 取消可能以包装异常的形式抛出（例如 SQLException by Statement.cancel()），通过 handle 反向判别 + if (handle.cancelled.get()) + { + val rows = handle.rowsAtStop + taskLog.warn("Local executor task stopped by user (via JDBC cancel): rows≈{} task={}", rows, request.taskName) + response.count = if (rows > Int.MAX_VALUE.toLong()) Int.MAX_VALUE else rows.toInt() + response.successful = false + response.state = RunState.STOPPED + response.message = "Stopped by user" + } + else + { + taskLog.error("Local executor failed", ex) + response.successful = false + response.state = RunState.FAILURE + response.message = ex.message + } } finally { + if (request.taskName.isNotBlank()) + { + runningTasks.remove(request.taskName, handle) + } loggerExecutor?.destroy() } return response } + /** 取消传播专用异常，携带已处理行数便于上游记录 */ + private class TaskCancelledException(val processed: Long) : RuntimeException("Task cancelled by user") + private fun newTaskLogger(request: ExecutorRequest): LoggerExecutor<*>? { val workHome = request.workHome @@ -143,7 +194,42 @@ class LocalExecutorService : ExecutorService override fun stop(request: ExecutorRequest): ExecutorResponse { - return ExecutorResponse(false, true, RunState.SUCCESS, null) + val taskName = request.taskName + if (taskName.isBlank()) + { + return ExecutorResponse(false, false, RunState.FAILURE, "taskName is required") + } + val handle = runningTasks[taskName] + if (handle == null) + { + return ExecutorResponse(false, false, RunState.FAILURE, "Task [ $taskName ] is not running on this node") + } + // 设标志位 + 写日志：纯内存操作，立即完成 + handle.cancelled.set(true) + log.info("Cancel requested for task [ {} ]", taskName) + handle.taskLog?.warn("Stop requested by user for task [ {} ]", taskName) + // Statement.cancel() 实现里通常会新开一个 JDBC 连接发 KILL QUERY， + // 如果源 DB 网络异常会阻塞 HTTP 请求线程。所以放到独立线程，不让 HTTP 等 + val stmt = handle.sourceStatement + if (stmt != null) + { + cancelExecutor.submit { + try + { + if (!stmt.isClosed) + { + stmt.cancel() + handle.taskLog?.warn("Source JDBC statement cancelled for task [ {} ]", taskName) + } + } + catch (ex: Exception) + { + log.warn("Cancel source statement failed: {}", ex.message) + handle.taskLog?.warn("Cancel source statement failed: {}", ex.message) + } + } + } + return ExecutorResponse(false, true, RunState.STOPPED, null) } /** @@ -202,7 +288,8 @@ class LocalExecutorService : ExecutorService batchSize: Int, taskLog: Logger, totalCount: Long, - progressListener: ExecutorProgressListener? + progressListener: ExecutorProgressListener?, + handle: TaskHandle ): Long { taskLog.info( @@ -225,8 +312,17 @@ class LocalExecutorService : ExecutorService taskLog.info("Streaming sync: source returned headers={}", headers) } + override fun onStatement(statement: java.sql.Statement) + { + handle.sourceStatement = statement + } + override fun onRow(row: List) { + if (handle.cancelled.get()) + { + throw TaskCancelledException(written) + } val projected = ArrayList(sourceKeys.size) for (key in sourceKeys) { @@ -245,10 +341,17 @@ class LocalExecutorService : ExecutorService written, writer.writtenCount(), "%.1f".format(seconds), rps ) progressListener?.onProgress(writer.writtenCount(), totalCount) + // 顺手记录最近一次进度行数，给 stop 后 catch 取整时使用 + handle.rowsAtStop = writer.writtenCount() } } }) } + // 某些驱动在 Statement.cancel() 后直接让 rs.next() 返回 false（不抛异常），需要在循环结束后再判一次 + if (handle.cancelled.get()) + { + throw TaskCancelledException(written) + } val totalSeconds = (System.nanoTime() - startNanos) / 1_000_000_000.0 taskLog.info( "Streaming sync done: rows={} committed={} elapsed={}s", @@ -273,7 +376,8 @@ class LocalExecutorService : ExecutorService batchSize: Int, taskLog: Logger, totalCount: Long, - progressListener: ExecutorProgressListener? + progressListener: ExecutorProgressListener?, + handle: TaskHandle ): Long { taskLog.info("Legacy sync start: target=`{}`.`{}` batchSize={}", database, table, batchSize) @@ -300,6 +404,10 @@ class LocalExecutorService : ExecutorService writer.use { writer -> for (item in rows) { + if (handle.cancelled.get()) + { + throw TaskCancelledException(written) + } val node = item as? ObjectNode ?: continue val projected = ArrayList(sourceKeys.size) for (key in sourceKeys) @@ -325,6 +433,10 @@ class LocalExecutorService : ExecutorService val batch = ArrayList(batchSize) for (item in rows) { + if (handle.cancelled.get()) + { + throw TaskCancelledException(written) + } val node = item as? ObjectNode ?: continue val sqlColumns = ArrayList(originColumns.size) for (col in originColumns) @@ -448,5 +560,17 @@ class LocalExecutorService : ExecutorService private const val DEFAULT_FETCH_SIZE = 1000 private const val DEFAULT_BATCH_SIZE = 1000 private const val PROGRESS_INTERVAL = 1_000L + + // 进程内活跃任务表。多个并发 sync 共享同一 LocalExecutorService 实例， + // 用 taskName 唯一索引；stop() 通过 taskName 查到 handle 后设置取消标志位 + private val runningTasks: ConcurrentHashMap = ConcurrentHashMap() + + // Statement.cancel() 可能会阻塞（驱动会新开连接发 KILL），放到独立线程跑避免拖住调用方 + private val cancelExecutor: java.util.concurrent.ExecutorService = + java.util.concurrent.Executors.newCachedThreadPool { r -> + val t = Thread(r, "local-executor-cancel") + t.isDaemon = true + t + } } } diff --git a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/common/RunState.kt b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/common/RunState.kt index a6400b7330..3cabb4db51 100644 --- a/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/common/RunState.kt +++ b/executor/datacap-executor-spi/src/main/kotlin/io/edurt/datacap/executor/common/RunState.kt @@ -7,5 +7,6 @@ enum class RunState { RUNNING, FAILURE, SUCCESS, + STOPPING, STOPPED } From 52030674e40e3d619c2703a995d4738383398994 Mon Sep 17 00:00:00 2001 From: qianmoQ Date: Thu, 21 May 2026 16:32:20 +0800 Subject: [PATCH 10/23] =?UTF-8?q?feat(dataset):=20=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E9=87=8D=E5=90=AF=E5=90=8E=E6=81=A2=E5=A4=8D=E6=9C=AA=E5=AE=8C?= =?UTF-8?q?=E7=BB=93=E5=90=8C=E6=AD=A5=E5=8E=86=E5=8F=B2=E4=B8=BA=20INTERR?= =?UTF-8?q?UPTED?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RunState 新增 INTERRUPTED：服务异常退出后留下的 RUNNING / CREATED / STOPPING 历史在下次启动时被标记为此状态 - DatasetHistoryRepository 加 findAllByStateIn(Collection) 供恢复扫描 - 新增 DatasetHistoryRecoveryRunner（CommandLineRunner，Order=10，早于 SchedulerRunner）： - 启动时批量把非终态历史改为 INTERRUPTED - message 带时间戳与原状态，便于审计 - UI 增加紫色标签 + 国际化 state.common.interrupted；DatasetHistory.vue hover-card 同时支持 FAILURE 与 INTERRUPTED 展示 message --- .../etc/conf/i18n/messages_en.properties | 1 + .../etc/conf/i18n/messages_zh-cn.properties | 1 + .../runner/DatasetHistoryRecoveryRunner.java | 70 +++++++++++++++++++ .../repository/DatasetHistoryRepository.java | 10 +++ core/datacap-ui/src/utils/common.ts | 4 ++ .../pages/admin/dataset/DatasetHistory.vue | 7 +- .../edurt/datacap/executor/common/RunState.kt | 8 ++- 7 files changed, 99 insertions(+), 2 deletions(-) create mode 100644 core/datacap-server/src/main/java/io/edurt/datacap/server/runner/DatasetHistoryRecoveryRunner.java diff --git a/configure/etc/conf/i18n/messages_en.properties b/configure/etc/conf/i18n/messages_en.properties index 6df5b948ca..6c1c689f43 100644 --- a/configure/etc/conf/i18n/messages_en.properties +++ b/configure/etc/conf/i18n/messages_en.properties @@ -14,6 +14,7 @@ state.common.success=Run Successful state.common.failure=Run Failed state.common.stop=Stopped state.common.stopping=Stopping +state.common.interrupted=Interrupted state.common.timeout=Run Timed Out state.common.queue=Queued ## Query i18n diff --git a/configure/etc/conf/i18n/messages_zh-cn.properties b/configure/etc/conf/i18n/messages_zh-cn.properties index 6059a8658b..a5c394d0c2 100644 --- a/configure/etc/conf/i18n/messages_zh-cn.properties +++ b/configure/etc/conf/i18n/messages_zh-cn.properties @@ -14,6 +14,7 @@ state.common.success=\u8FD0\u884C\u6210\u529F state.common.failure=\u8FD0\u884C\u5931\u8D25 state.common.stop=\u5DF2\u505C\u6B62 state.common.stopping=\u505C\u6B62\u4E2D +state.common.interrupted=\u5F02\u5E38\u4E2D\u65AD state.common.timeout=\u8FD0\u884C\u8D85\u65F6 state.common.queue=\u6392\u961F\u4E2D ## Query i18n diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/runner/DatasetHistoryRecoveryRunner.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/runner/DatasetHistoryRecoveryRunner.java new file mode 100644 index 0000000000..51e29a0195 --- /dev/null +++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/runner/DatasetHistoryRecoveryRunner.java @@ -0,0 +1,70 @@ +package io.edurt.datacap.server.runner; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import io.edurt.datacap.executor.common.RunState; +import io.edurt.datacap.service.entity.DatasetHistoryEntity; +import io.edurt.datacap.service.repository.DatasetHistoryRepository; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.time.DateFormatUtils; +import org.springframework.boot.CommandLineRunner; +import org.springframework.core.annotation.Order; +import org.springframework.stereotype.Service; + +import java.util.Arrays; +import java.util.Date; +import java.util.List; + +/** + * 启动时把上次服务异常退出留下的"未完结"同步历史标记为 INTERRUPTED。 + * 仅处理 RUNNING / CREATED / STOPPING 这些非终态。 + * 在 SchedulerRunner 之前执行（@Order 较小），避免被即将启动的定时任务"运行中"覆盖。 + * + * On startup, mark sync-history rows left in transient states by a prior crash as INTERRUPTED. + */ +@Slf4j +@Service +@Order(10) +@SuppressFBWarnings(value = {"EI_EXPOSE_REP2"}) +public class DatasetHistoryRecoveryRunner + implements CommandLineRunner +{ + private static final List TRANSIENT_STATES = Arrays.asList( + RunState.RUNNING, + RunState.CREATED, + RunState.STOPPING + ); + + private final DatasetHistoryRepository historyRepository; + + public DatasetHistoryRecoveryRunner(DatasetHistoryRepository historyRepository) + { + this.historyRepository = historyRepository; + } + + @Override + public void run(String... args) + { + try { + List stale = historyRepository.findAllByStateIn(TRANSIENT_STATES); + if (stale.isEmpty()) { + log.info("Dataset history recovery: no stale records."); + return; + } + String now = DateFormatUtils.format(System.currentTimeMillis(), "yyyy-MM-dd HH:mm:ss"); + String message = String.format("Marked as INTERRUPTED by server restart at %s (previous state was transient).", now); + Date updateTime = new Date(); + for (DatasetHistoryEntity history : stale) { + RunState previous = history.getState(); + history.setState(RunState.INTERRUPTED); + history.setMessage(message); + history.setUpdateTime(updateTime); + log.info("Recovering sync history [ {} ] previousState={} -> INTERRUPTED", history.getId(), previous); + } + historyRepository.saveAll(stale); + log.info("Dataset history recovery: {} stale record(s) marked INTERRUPTED.", stale.size()); + } + catch (Exception ex) { + log.error("Dataset history recovery failed", ex); + } + } +} diff --git a/core/datacap-service/src/main/java/io/edurt/datacap/service/repository/DatasetHistoryRepository.java b/core/datacap-service/src/main/java/io/edurt/datacap/service/repository/DatasetHistoryRepository.java index 4e289ea672..08dde9aeaa 100644 --- a/core/datacap-service/src/main/java/io/edurt/datacap/service/repository/DatasetHistoryRepository.java +++ b/core/datacap-service/src/main/java/io/edurt/datacap/service/repository/DatasetHistoryRepository.java @@ -1,13 +1,23 @@ package io.edurt.datacap.service.repository; +import io.edurt.datacap.executor.common.RunState; import io.edurt.datacap.service.entity.DataSetEntity; import io.edurt.datacap.service.entity.DatasetHistoryEntity; import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; import org.springframework.data.repository.PagingAndSortingRepository; +import java.util.Collection; +import java.util.List; + public interface DatasetHistoryRepository extends PagingAndSortingRepository { Page findAllByDatasetOrderByCreateTimeDesc(DataSetEntity dataSet, Pageable pageable); + + /** + * 启动恢复用：查出"未完结"的同步历史 + * Used by startup recovery to find sync history rows that did not reach a terminal state. + */ + List findAllByStateIn(Collection states); } diff --git a/core/datacap-ui/src/utils/common.ts b/core/datacap-ui/src/utils/common.ts index 498eccc1a1..3ca43d3ddc 100644 --- a/core/datacap-ui/src/utils/common.ts +++ b/core/datacap-ui/src/utils/common.ts @@ -33,6 +33,8 @@ const getColor = (origin: string): string => { return 'hsl(38 92% 50%)' case 'STOPPED': return '#17233d' + case 'INTERRUPTED': + return 'hsl(280 60% 50%)' case 'TIMEOUT': return 'hsl(47.9 95.8% 53.1%)' default: @@ -85,6 +87,8 @@ export function useUtil() return t('state.common.stopping') case 'STOPPED': return t('state.common.stop') + case 'INTERRUPTED': + return t('state.common.interrupted') case 'TIMEOUT': return t('state.common.timeout') case 'QUEUE': diff --git a/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue b/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue index 106f993672..bd11d155cb 100644 --- a/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue +++ b/core/datacap-ui/src/views/pages/admin/dataset/DatasetHistory.vue @@ -8,7 +8,7 @@