Skip to content

Refactor LocalExecutorService and enhance testing coverage #967

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
qianmoQ wants to merge 17 commits into
base: dev
Choose a base branch
from
Open

Refactor LocalExecutorService and enhance testing coverage #967

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
c440c75
test(executor): add characterization tests for local executor sync
qianmoQ Sep 4, 2026
ab76f13
refactor(executor): split LocalExecutorService into sync strategies
qianmoQ Sep 4, 2026
2c349d8
fix(executor): report committed row count when a sync is stopped
qianmoQ Sep 4, 2026
ded497f
feat(executor): enforce optional whole-task timeout in local executor
qianmoQ Sep 4, 2026
da69880
refactor(executor): drop hand-built SQL fallback from local executor
qianmoQ Sep 4, 2026
4565336
test(executor): add PostgreSQL Testcontainers end-to-end sync tests
qianmoQ Sep 4, 2026
0e9b894
test(executor): pin seatunnel request-to-command mapping
qianmoQ Sep 4, 2026
00faa07
refactor(executor): split ExecutorRequest into common fields and options
qianmoQ Sep 4, 2026
0d5fadb
feat(ui): add ant-design-vue foundation alongside view-shadcn-ui
qianmoQ Sep 5, 2026
144a626
refactor(ui): migrate language switcher to ant-design-vue
qianmoQ Sep 5, 2026
7086da3
refactor(ui): migrate error pages to ant-design-vue
qianmoQ Sep 5, 2026
0380d6e
refactor(ui): migrate sign-in page to ant-design-vue
qianmoQ Sep 5, 2026
4da644d
refactor(ui): migrate sign-up page to ant-design-vue
qianmoQ Sep 5, 2026
d73f2f0
refactor(ui): migrate user list page to ant-design-vue + composition API
qianmoQ Sep 5, 2026
06650ae
refactor(ui): convert error pages to script setup composition
qianmoQ Sep 5, 2026
a2f0ddc
refactor(ui): convert auth pages to script setup composition
qianmoQ Sep 5, 2026
8a829e0
refactor(ui): migrate user-role dialog to ant-design-vue + compositio…
qianmoQ Sep 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 13 additions & 22 deletions ...tacap-service/src/main/java/io/edurt/datacap/service/service/impl/DataSetServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,8 @@
import io.edurt.datacap.common.sql.configure.SqlOrder;
import io.edurt.datacap.common.sql.configure.SqlType;
import io.edurt.datacap.executor.ExecutorService;
import io.edurt.datacap.executor.common.RunEngine;
import io.edurt.datacap.executor.common.RunMode;
import io.edurt.datacap.executor.common.RunProtocol;
import io.edurt.datacap.executor.common.RunState;
import io.edurt.datacap.executor.common.RunWay;
import io.edurt.datacap.executor.configure.ExecutorConfigure;
import io.edurt.datacap.executor.configure.ExecutorRequest;
import io.edurt.datacap.executor.configure.ExecutorResponse;
Expand Down Expand Up @@ -1229,28 +1226,21 @@ private DataSetEntity syncData(DataSetEntity entity, java.util.concurrent.Execut
log.warn("Serialize effective executor configure failed: {}", ex.getMessage());
}
historyRepository.save(history);
int fetchSize = parseIntOrDefault(executorCfg.get("fetchSize"), 1000);
int batchSize = parseIntOrDefault(executorCfg.get("batchSize"), 1000);
boolean preCount = Boolean.parseBoolean(executorCfg.getOrDefault("preCount", "false"));
// timeout 由各 executor 自己的配置决定：LocalExecutor schema 默认 0（不限时），
// 未声明该字段的 executor（如 Seatunnel）回退到 600 秒，保持原有行为
int timeout = parseIntOrDefault(executorCfg.get("timeout"), 600);
ExecutorRequest request = new ExecutorRequest(
taskName,
entity.getUser().getUsername(),
input,
output,
executorCfg.get("home"),
workHome,
this.pluginManager,
600,
RunWay.valueOf(executorCfg.getOrDefault("way", "LOCAL")),
RunMode.valueOf(executorCfg.getOrDefault("mode", "CLIENT")),
executorCfg.get("startScript"),
RunEngine.valueOf(executorCfg.getOrDefault("engine", "SPARK")),
null,
fetchSize,
batchSize,
preCount,
null
output
);
request.setWorkHome(workHome);
request.setTimeout(timeout);
request.setPluginManager(this.pluginManager);
// 把该 executor 的 effective 配置整体作为 options，各执行器各取所需：
// Local 读 fetchSize/batchSize/preCount；Seatunnel 读 home/startScript/way/mode/engine
request.setOptions(executorCfg);

// 进度回调：每个 batch 写完后更新 datacap_dataset_history 以及数据集自身的 totalRows / totalSize
// totalRows 直接用已写入计数；totalSize 必须查 ClickHouse system.parts，开销大且不可控，
Expand Down Expand Up @@ -1355,8 +1345,9 @@ 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) {
// STOPPED（用户主动停止）与 TIMEOUT（超时取消）都是终态：已把最终状态写入 history，
// 无需当作异常抛出 / 也不刷 table metadata
if (response.getState() == RunState.STOPPED || response.getState() == RunState.TIMEOUT) {
return;
}
Preconditions.checkArgument(response.getSuccessful(), response.getMessage());
Expand Down
19 changes: 6 additions & 13 deletions ...acap-service/src/main/java/io/edurt/datacap/service/service/impl/WorkflowServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,7 @@
import io.edurt.datacap.common.response.CommonResponse;
import io.edurt.datacap.common.utils.NullAwareBeanUtils;
import io.edurt.datacap.executor.ExecutorService;
import io.edurt.datacap.executor.common.RunEngine;
import io.edurt.datacap.executor.common.RunMode;
import io.edurt.datacap.executor.common.RunState;
import io.edurt.datacap.executor.common.RunWay;
import io.edurt.datacap.executor.configure.ExecutorConfigure;
import io.edurt.datacap.executor.configure.ExecutorRequest;
import io.edurt.datacap.executor.configure.ExecutorResponse;
Expand Down Expand Up @@ -155,22 +152,18 @@ public CommonResponse<WorkflowEntity> saveOrUpdate(BaseRepository<WorkflowEntity
plugin.getName(),
plugin.configures()
);
String executorHome = executorCfg.get("home");
log.debug("Executor home directory: {}", executorHome);
log.debug("Executor home directory: {}", executorCfg.get("home"));

ExecutorRequest request = new ExecutorRequest(
configure.getWork(),
executorHome,
configure.getCode(),
user.getUsername(),
form,
to,
RunMode.valueOf(executorCfg.getOrDefault("mode", "CLIENT")),
RunWay.valueOf(executorCfg.getOrDefault("way", "LOCAL")),
executorCfg.get("startScript"),
RunEngine.valueOf(executorCfg.getOrDefault("engine", "SPARK")),
transform
to
);
request.setWorkHome(configure.getWork());
request.setTransform(transform);
// executor 专属项（home/startScript/way/mode/engine 等）整体作为 options 传入
request.setOptions(executorCfg);
log.info("Created executor request for workflow: {}", configure.getCode());

Timestamp start = new Timestamp(System.currentTimeMillis());
Expand Down
9 changes: 9 additions & 0 deletions core/datacap-ui/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
"preview": "vite preview"
},
"dependencies": {
"@ant-design/icons-vue": "^7.0.1",
"@antv/x6": "^2.18.1",
"@fortawesome/fontawesome-svg-core": "^6.5.1",
"@fortawesome/free-solid-svg-icons": "^6.5.1",
Expand All @@ -22,6 +23,7 @@
"ag-grid-community": "^31.3.4",
"ag-grid-vue3": "^31.3.4",
"ansi_up": "^6.0.2",
"ant-design-vue": "^4.2.6",
"axios": "^1.7.4",
"clsx": "^2.1.0",
"lodash": "^4.17.21",
Expand All @@ -45,5 +47,12 @@
"typescript": "^5.2.2",
"vite": "^5.4.8",
"vue-tsc": "^1.8.27"
},
"pnpm": {
"onlyBuiltDependencies": [
"esbuild",
"vue-demi",
"core-js"
]
}
}
Loading
Loading