jerryshao commented on code in PR #13250:
URL: https://github.com/apache/gravitino/pull/13250#discussion_r4059004174
##########
clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java:
##########
@@ -1714,14 +1714,17 @@ public JobHandle runJob(String jobTemplateName,
Map<String, String> jobConf)
}
@Override
- public JobHandle getJob(String jobId) throws NoSuchJobException {
+ public JobHandle getJob(String jobId, boolean includeOutput) throws
NoSuchJobException {
Preconditions.checkArgument(StringUtils.isNotBlank(jobId), "job id must
not be null or empty");
+ Map<String, String> params =
+ includeOutput ? ImmutableMap.of("includeOutput", "true") :
Collections.emptyMap();
Review Comment:
[Nit] The Java client only ever sends `includeOutput`; the
`outputMaxLines`/`outputMaxBytes` query parameters added on the server
(`JobOperations.java:386-387`) and documented in
`docs/manage-jobs-in-gravitino.md:269-278` are unreachable from both official
clients (the Python client does the same at `gravitino_metalake.py:602`). A
caller who wants the "quick check that doesn't need the full 1000 lines" the
docs describe has to hand-build the URL.
Not blocking - just worth either plumbing the two caps through
`SupportsJobs#getJob` as well, or noting in the docs that they are REST-only
for now.
Verified by: reading `GravitinoMetalake.java:1717-1731`,
`gravitino_metalake.py:579-610` and `JobOperations.java:380-400` at `f41df58`.
##########
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:
[Question] Everything else on this path is careful to degrade to empty
output rather than fail - the `JobExecutor` javadoc, `JobManager.getJob`'s
comment, and the `!file.exists()` check on line 432 all say so. Here an
`IOException` becomes a `RuntimeException`, which `JobExceptionHandler` maps to
a 500.
`JobManager.cleanUpStagingDirs()` (`JobManager.java:795-843`) deletes the
job entity and then `FileUtils.deleteDirectory(jobStagingDir)`. A `getJob(...,
includeOutput=true)` that read the entity just before that delete, and reaches
this open just after, hits `FileNotFoundException` and returns 500 rather than
the empty output the rest of the design intends. The window is narrow, but it
is a scheduled background task, not an unlikely event.
Would it be worth catching `FileNotFoundException` specifically and
returning `ImmutableList.of()`, keeping the hard failure for genuine I/O errors?
Verified by: reading `LocalJobExecutor.java:431-454` and
`JobManager.cleanUpStagingDirs()` at `JobManager.java:795-843`.
##########
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:
[Important] When the byte window lands entirely inside a single line that
**is** newline-terminated, this returns an empty list instead of the truncated
tail.
`windowStartsAtLineBoundary` is false (the byte before the window is not
`\n`), so this branch drops everything up to and including the first `\n`. For
an oversized final line that ends with a newline, that `\n` is the *last*
character of the window, so `content` becomes `""` and lines 479-481 return
`ImmutableList.of()`. The comment just above (471-474) states the intended
behaviour - "keep it as-is rather than discarding it, since a truncated line is
more useful for debugging than silently returning nothing" - but that only
holds for a file with no trailing newline.
Concretely, with `maxBytes = 1024`:
- `'x' * 4096` (no trailing newline) -> 1 line, tail preserved. This is what
`TestLocalJobExecutor.java:377` writes.
- `'x' * 4096 + '\n'` -> `[]`.
- `"error: something failed\nstack frame 1\nstack frame 2\n" + 'y' * 4096 +
'\n'` -> `[]`.
So a job whose last line is a big one-line JSON/base64 blob - exactly the
shape that motivated the byte cap - reports *no output at all*, and the caller
cannot distinguish that from "the job printed nothing". Every stdout-producing
shell builtin (`echo`, `printf '...\n'`, `jq -c`) terminates its last line, so
this is the common case, not the exotic one.
Suggested fix: only drop the leading partial line when something survives
it, e.g. compute `firstNewline` and keep the original `content` when
`firstNewline == content.length() - 1` (or when the result would be empty), so
an oversized line still comes back truncated.
Verified by: reading `LocalJobExecutor.java:431-493` at `f41df58`, then
extracting `readLastLines` into a standalone class and running it against the
three files above (results as listed).
##########
core/src/test/java/org/apache/gravitino/job/local/TestLocalJobExecutor.java:
##########
@@ -259,6 +266,236 @@ public void
testSubmitSparkJobRejectedWhenSparkSubmitIsNotAvailable() throws IOE
}
}
+ @Test
+ public void testGetJobOutputSuccessfully() throws IOException {
+ Map<String, String> jobConf =
+ ImmutableMap.of(
+ "arg1", "value1",
+ "arg2", "success",
+ "var", "value3");
+
+ JobTemplate template =
+ JobManager.createRuntimeJobTemplate(jobTemplateEntity, jobConf,
workingDir);
+
+ String jobId = jobExecutor.submitJob(template);
+ Awaitility.await()
+ .atMost(3, TimeUnit.MINUTES)
+ .until(() -> jobExecutor.getJobStatus(jobId) ==
JobHandle.Status.SUCCEEDED);
+
+ List<String> stdout = jobExecutor.getJobStdout(jobId, 1000,
DEFAULT_TEST_MAX_BYTES);
+ Assertions.assertEquals(6, stdout.size());
+ Assertions.assertEquals("starting test test job", stdout.get(0));
+ Assertions.assertEquals("in common script", stdout.get(1));
+ Assertions.assertTrue(stdout.get(2).startsWith("Submitting job with
name:"));
+ Assertions.assertEquals("value1", stdout.get(3));
+ Assertions.assertEquals("success", stdout.get(4));
+ Assertions.assertEquals("value3", stdout.get(5));
+
+ // The test script never writes to stderr.
+ Assertions.assertEquals(
+ Collections.emptyList(), jobExecutor.getJobStderr(jobId, 1000,
DEFAULT_TEST_MAX_BYTES));
+
+ // The full output has 6 lines; only the last 3 should be returned when
capped.
+ Assertions.assertEquals(
+ ImmutableList.of("value1", "success", "value3"),
+ jobExecutor.getJobStdout(jobId, 3, DEFAULT_TEST_MAX_BYTES));
+ }
+
+ @Test
+ public void testGetJobOutputForUnknownJobReturnsEmpty() {
+ // A job unknown to this executor - whether it never existed here, or its
bookkeeping has
+ // expired/been lost - reports empty output rather than throwing: the job
entity itself may
+ // still exist, and querying its output must not turn that into an error.
+ Assertions.assertEquals(
+ Collections.emptyList(),
+ jobExecutor.getJobStdout("no-such-job", 100, DEFAULT_TEST_MAX_BYTES));
+ Assertions.assertEquals(
+ Collections.emptyList(),
+ jobExecutor.getJobStderr("no-such-job", 100, DEFAULT_TEST_MAX_BYTES));
+ }
+
+ @Test
+ public void testGetJobOutputForQueuedJobReturnsEmpty() throws IOException {
+ LocalJobExecutor exec = new LocalJobExecutor();
+ exec.initialize(ImmutableMap.of(LocalJobExecutorConfigs.MAX_RUNNING_JOBS,
"1"));
+
+ File workingDirA =
Files.createTempDirectory("gravitino-test-local-job-executor-a").toFile();
+ File workingDirB =
Files.createTempDirectory("gravitino-test-local-job-executor-b").toFile();
+ try {
+ Map<String, String> jobConf =
+ ImmutableMap.of(
+ "arg1", "value1",
+ "arg2", "success",
+ "var", "value3");
+
+ // Submit two jobs to a single-threaded executor - the second one stays
QUEUED until the
+ // first (which sleeps for a few seconds) finishes.
+ JobTemplate templateA =
+ JobManager.createRuntimeJobTemplate(jobTemplateEntity, jobConf,
workingDirA);
+ JobTemplate templateB =
+ JobManager.createRuntimeJobTemplate(jobTemplateEntity, jobConf,
workingDirB);
+ exec.submitJob(templateA);
+ String jobIdB = exec.submitJob(templateB);
+
+ Assertions.assertEquals(JobHandle.Status.QUEUED,
exec.getJobStatus(jobIdB));
+ Assertions.assertEquals(
+ Collections.emptyList(), exec.getJobStdout(jobIdB, 100,
DEFAULT_TEST_MAX_BYTES));
+ Assertions.assertEquals(
+ Collections.emptyList(), exec.getJobStderr(jobIdB, 100,
DEFAULT_TEST_MAX_BYTES));
+
+ Awaitility.await()
+ .atMost(3, TimeUnit.MINUTES)
+ .until(() -> exec.getJobStatus(jobIdB) ==
JobHandle.Status.SUCCEEDED);
+ } finally {
+ exec.close();
+ FileUtils.deleteDirectory(workingDirA);
+ FileUtils.deleteDirectory(workingDirB);
+ }
+ }
+
+ @Test
+ public void testGetJobOutputWithOversizedSingleLineIsBoundedByMaxBytes()
throws IOException {
+ Map<String, String> jobConf =
+ ImmutableMap.of(
+ "arg1", "value1",
+ "arg2", "success",
+ "var", "value3");
+
+ JobTemplate template =
+ JobManager.createRuntimeJobTemplate(jobTemplateEntity, jobConf,
workingDir);
+
+ String jobId = jobExecutor.submitJob(template);
+ Awaitility.await()
+ .atMost(3, TimeUnit.MINUTES)
+ .until(() -> jobExecutor.getJobStatus(jobId) ==
JobHandle.Status.SUCCEEDED);
+
+ // Overwrite the captured output with a single line far larger than the
byte window, with no
+ // trailing newline - the pathological case a byte-bounded tail read must
stay safe against
+ // (a naive line-oriented reverse reader can degrade badly on content
shaped like this).
+ int maxBytes = 1024;
+ String oversizedLine = StringUtils.repeat('x', maxBytes * 4);
+ FileUtils.writeStringToFile(new File(workingDir, "output.log"),
oversizedLine, "UTF-8");
Review Comment:
[Nit] This writes the oversized line without a trailing newline, which is
the single file shape that survives the partial-line drop at line 476. Two
variants would pin down the behaviour the method's own comment promises, and
both currently return `[]`:
- the same line with a trailing `\n`;
- a few ordinary lines followed by an oversized, newline-terminated final
line (asserting the truncated tail of that last line is returned).
Verified by: running the extracted `readLastLines` logic against both shapes
with `maxBytes = 1024`; see the comment on `LocalJobExecutor.java:476`.
--
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]
