jerryshao commented on code in PR #13250:
URL: https://github.com/apache/gravitino/pull/13250#discussion_r4059082032
##########
core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java:
##########
@@ -377,9 +403,93 @@ void cleanupJobStatus() {
jobStatus
.entrySet()
.removeIf(
- entry ->
- entry.getValue().getRight() != UNEXPIRED_TIME_IN_MS
- && (currentTime - entry.getValue().getRight()) >=
jobStatusKeepTimeInMs);
+ entry -> {
+ boolean expired =
+ entry.getValue().getRight() != UNEXPIRED_TIME_IN_MS
+ && (currentTime - entry.getValue().getRight()) >=
jobStatusKeepTimeInMs;
+ if (expired) {
+ jobWorkingDirs.remove(entry.getKey());
+ }
+ return expired;
+ });
+ }
+ }
+
+ private List<String> getJobOutput(String jobId, String fileName, int
maxLines, int maxBytes) {
+ File workingDir = getWorkingDir(jobId);
+ if (workingDir == null) {
+ return ImmutableList.of();
+ }
+ return readLastLines(new File(workingDir, fileName), maxLines, maxBytes);
+ }
+
+ @Nullable
+ private File getWorkingDir(String jobId) {
+ return jobWorkingDirs.get(jobId);
+ }
+
+ private List<String> readLastLines(File file, int maxLines, int maxBytes) {
+ if (!file.exists()) {
+ // The job hasn't started (or hasn't produced this stream) yet.
+ return ImmutableList.of();
+ }
+
+ long fileLength = file.length();
+ int windowSize = (int) Math.min(fileLength, maxBytes);
+ long startOffset = fileLength - windowSize;
+
+ // Read one extra leading byte (when available) so we can tell whether the
window's first
+ // line is already complete - i.e. the file byte immediately before the
window is itself a
+ // line terminator - rather than always assuming it's a partial line and
discarding it.
+ long readOffset = Math.max(0, startOffset - 1);
+ byte[] probeWindow = new byte[(int) (fileLength - readOffset)];
+ try (RandomAccessFile raf = new RandomAccessFile(file, "r")) {
+ raf.seek(readOffset);
+ raf.readFully(probeWindow);
+ } catch (IOException e) {
+ // An I/O failure while reading an existing file is unexpected (unlike
the file simply not
+ // existing yet, handled above) and must not be silently reported as "no
output" - that
+ // would be actively misleading for the debugging use case this method
exists for.
+ throw new RuntimeException("Failed to read job output file: " + file, e);
}
+
+ boolean windowStartsAtLineBoundary = startOffset == 0 || probeWindow[0] ==
'\n';
+ int contentStart = startOffset == 0 ? 0 : 1;
+ // The window may start mid-character if the file byte at contentStart
happens to be a UTF-8
+ // continuation byte - skip forward to the next character boundary so the
decoded content
+ // never begins with a corrupted replacement character. A '\n' byte never
appears inside a
+ // multi-byte UTF-8 character, so this can't skip past a real line
boundary.
+ while (contentStart < probeWindow.length && (probeWindow[contentStart] &
0xC0) == 0x80) {
+ contentStart++;
+ }
+
+ String content =
+ new String(
+ probeWindow, contentStart, probeWindow.length - contentStart,
StandardCharsets.UTF_8);
+ if (!windowStartsAtLineBoundary) {
+ // The window starts mid-file and the preceding byte isn't a line
terminator, so its first
+ // line may be a partial line whose true beginning fell outside the
window - drop up to and
+ // including the first newline. If none is found, the window is entirely
a single oversized
+ // line; keep it as-is rather than discarding it, since a truncated line
is more useful for
+ // debugging than silently returning nothing.
+ int firstNewline = content.indexOf('\n');
+ content = firstNewline >= 0 ? content.substring(firstNewline + 1) :
content;
Review Comment:
Confirmed and fixed at cbdf957. You're right — the only `\n` in the window
terminating the oversized line itself was being mistaken for "there's more
content after this boundary," so it got dropped along with everything before it.
Fixed by only dropping the leading fragment when the found newline isn't the
window's last character (i.e. something genuinely follows it):
```java
int firstNewline = content.indexOf('\n');
if (firstNewline >= 0 && firstNewline < content.length() - 1) {
content = content.substring(firstNewline + 1);
}
```
Added the two test variants you suggested (oversized line ending in `\n`,
and ordinary lines followed by an oversized newline-terminated final line) —
both now correctly return the truncated tail instead of `[]`.
##########
core/src/main/java/org/apache/gravitino/job/local/LocalJobExecutor.java:
##########
@@ -377,9 +403,93 @@ void cleanupJobStatus() {
jobStatus
.entrySet()
.removeIf(
- entry ->
- entry.getValue().getRight() != UNEXPIRED_TIME_IN_MS
- && (currentTime - entry.getValue().getRight()) >=
jobStatusKeepTimeInMs);
+ entry -> {
+ boolean expired =
+ entry.getValue().getRight() != UNEXPIRED_TIME_IN_MS
+ && (currentTime - entry.getValue().getRight()) >=
jobStatusKeepTimeInMs;
+ if (expired) {
+ jobWorkingDirs.remove(entry.getKey());
+ }
+ return expired;
+ });
+ }
+ }
+
+ private List<String> getJobOutput(String jobId, String fileName, int
maxLines, int maxBytes) {
+ File workingDir = getWorkingDir(jobId);
+ if (workingDir == null) {
+ return ImmutableList.of();
+ }
+ return readLastLines(new File(workingDir, fileName), maxLines, maxBytes);
+ }
+
+ @Nullable
+ private File getWorkingDir(String jobId) {
+ return jobWorkingDirs.get(jobId);
+ }
+
+ private List<String> readLastLines(File file, int maxLines, int maxBytes) {
+ if (!file.exists()) {
+ // The job hasn't started (or hasn't produced this stream) yet.
+ return ImmutableList.of();
+ }
+
+ long fileLength = file.length();
+ int windowSize = (int) Math.min(fileLength, maxBytes);
+ long startOffset = fileLength - windowSize;
+
+ // Read one extra leading byte (when available) so we can tell whether the
window's first
+ // line is already complete - i.e. the file byte immediately before the
window is itself a
+ // line terminator - rather than always assuming it's a partial line and
discarding it.
+ long readOffset = Math.max(0, startOffset - 1);
+ byte[] probeWindow = new byte[(int) (fileLength - readOffset)];
+ try (RandomAccessFile raf = new RandomAccessFile(file, "r")) {
+ raf.seek(readOffset);
+ raf.readFully(probeWindow);
+ } catch (IOException e) {
+ // An I/O failure while reading an existing file is unexpected (unlike
the file simply not
+ // existing yet, handled above) and must not be silently reported as "no
output" - that
+ // would be actively misleading for the debugging use case this method
exists for.
+ throw new RuntimeException("Failed to read job output file: " + file, e);
Review Comment:
Good catch, fixed at cbdf957. Added a dedicated `catch
(FileNotFoundException e)` before the general `IOException` handler, degrading
to an empty list — consistent with the `!file.exists()` check right above it,
since this is the same "output no longer available" situation the cleanup race
just surfaces later than that initial check.
Added a test that simulates the race deterministically (delete the file and
replace it with a directory of the same name between the exists() check passing
and the actual read) rather than relying on real timing.
--
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]
