HyukjinKwon commented on code in PR #59328:
URL: https://github.com/apache/spark/pull/59328#discussion_r4266475713
##########
core/src/test/scala/org/apache/spark/deploy/history/EventLogFileCompactorSuite.scala:
##########
@@ -202,6 +205,107 @@ class EventLogFileCompactorSuite extends SparkFunSuite {
}
}
+ test("SPARK-60110: Don't compact files if some lines exceed the line-length
limit") {
+ withTempDir { dir =>
+ val fs = new Path(dir.getAbsolutePath).getFileSystem(hadoopConf)
+
+ // job 1 is finished
+ val stage1 = createStage(1, createRddsWithId(1 to 2), Nil)
+ val tasks1 = createTasks(4, Array("exec1"),
0L).map(createTaskStartEvent(_, 1, 0))
+
+ // job 2 is still live, and its job start line exceeds the line-length
limit
+ val stage2 = createStage(2, createRddsWithId(3 to 4), Nil)
+ val tasks2 = createTasks(4, Array("exec1"),
4L).map(createTaskStartEvent(_, 2, 0))
+ val maxLineLength = 8 * 1024
+ val props = new Properties()
+ props.setProperty("large", "x" * maxLineLength)
+ val jobStart2 = SparkListenerJobStart(2, 0, Seq(stage2), props)
+ val stageSubmitted2 = SparkListenerStageSubmitted(stage2)
+
+ // 1~3 are candidates to compact, 4~6 are dummies to ensure max files to
retain
+ val fileStatuses = writeEventsToRollingWriter(fs, "app", dir, sparkConf,
hadoopConf,
+ Seq(SparkListenerJobStart(1, 0, Seq(stage1)),
SparkListenerStageSubmitted(stage1)) ++
+ tasks1,
+ Seq(jobStart2, stageSubmitted2) ++ tasks2,
+ Seq(SparkListenerJobEnd(1, 0, JobSucceeded)),
+ testEvent,
+ testEvent,
+ testEvent)
+
+ // Replay skips job 2's start, so the filters would treat job 2 as
finished and drop
Review Comment:
**Nit (P3):** Without the gate, only job 2's `SparkListenerJobStart` would
be dropped, not all of its events. `BasicEventFilterBuilder.onStageSubmitted`
registers stage 2 even though the job start was skipped, and `onTaskStart` then
adds `tasks2` to it. `BasicEventFilter` would keep `stageSubmitted2` and
`tasks2` and reject only `jobStart2`, because `liveJobs` lacks job 2. The
assertion still catches a regression, but the comment gives the wrong picture
of the loss the gate prevents. Maybe something like: "Replay skips job 2's
start, so the filters would not know that job 2 is live and the rewrite would
drop its job start, leaving its stage and task events without the job.
Compaction must not proceed."
**Verification:**
- **Inspection:** The corrected comment matches
BasicEventFilterBuilder.onStageSubmitted/onTaskStart and JobEventFilter: stage
2 and tasks2 stay live without the job start, and only JobStart(2) is rejected.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
