jerryshao commented on code in PR #13250:
URL: https://github.com/apache/gravitino/pull/13250#discussion_r4059082731
##########
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:
Added both at cbdf957:
`testGetJobOutputWithOversizedSingleLineEndingInNewlineIsTruncatedNotEmpty` and
`testGetJobOutputWithOrdinaryLinesFollowedByOversizedFinalLine`. Both would
have caught the bug you flagged above before this fix — they now pass against
the corrected `readLastLines`.
--
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]
