davidzollo commented on code in PR #18492:
URL:
https://github.com/apache/dolphinscheduler/pull/18492#discussion_r4015984246
##########
dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelTask.java:
##########
@@ -121,12 +127,60 @@ public void trackApplicationStatus() throws TaskException
{
public void cancelApplication() throws TaskException {
// cancel process
try {
- shellCommandExecutor.cancelApplication();
+ cancelShellProcess();
} catch (Exception e) {
throw new TaskException("cancel application error", e);
}
}
+ protected void cancelShellProcess() throws Exception {
+ shellCommandExecutor.cancelApplication();
+ }
+
+ /**
+ * Extract SeaTunnel Engine job id from a log line.
+ *
+ * @param line log line
+ * @return job id or null
+ */
+ public static String findSeaTunnelJobId(String line) {
+ if (line == null) {
+ return null;
+ }
+ Matcher matcher = SEATUNNEL_JOB_ID_PATTERN.matcher(line);
+ if (matcher.find()) {
+ return matcher.group(1);
+ }
+ return null;
+ }
+
+ /**
+ * Parse SeaTunnel Engine job ids from the task log file.
+ */
+ protected List<String> findSeaTunnelJobIdsFromLog() {
+ String logPath = taskRequest.getLogPath();
+ if (logPath == null || logPath.isEmpty()) {
+ return Collections.emptyList();
+ }
+ File logFile = new File(logPath);
+ if (!logFile.exists() || !logFile.isFile()) {
+ return Collections.emptyList();
+ }
+ Set<String> jobIds = new HashSet<>();
+ try {
+ for (String line : Files.readAllLines(Paths.get(logPath),
StandardCharsets.UTF_8)) {
+ String jobId = findSeaTunnelJobId(line);
+ if (jobId != null) {
+ jobIds.add(jobId);
+ }
+ }
Review Comment:
readAllLines method can OOM the worker.
findSeaTunnelJobIdsFromLog calls it after every Zeta task exits, batch or
streaming, success or failure. Streaming tasks are exactly the target of this
PR and run for weeks; the client prints a job-progress table every 60s by
default, local deploy mode dumps all engine logs to stdout, and Console sinks
print every row. A multi-hundred-MB log becomes several times that on the heap.
The worker runs with a 4g default heap and -XX:+ExitOnOutOfMemoryError, so one
such task takes down the worker and every task on it. DS's own precedent for
the same job,
[LogUtils.getAppIdsFromLogFile](https://github.com/apache/dolphinscheduler/blob/9839c418c1a6d7f2a8c3395552edcd99c2de1b37/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/LogUtils.java#L163-L183),
streams with Files.lines.
Minimum fix: stream line by line and stop at the first match. Better fix:
capture the id from the stdout stream while the task runs instead of re-reading
the file at all.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
