dongjoon-hyun commented on code in PR #59328:
URL: https://github.com/apache/spark/pull/59328#discussion_r4236798149
##########
core/src/test/scala/org/apache/spark/deploy/history/EventLogFileCompactorSuite.scala:
##########
@@ -320,6 +425,20 @@ class EventLogFileCompactorSuite extends SparkFunSuite {
originalFiles.foreach { status => assert(fs.exists(status.getPath)) }
}
+ private def writeLinesToRollingWriter(
Review Comment:
nit. This is nearly the same as
`EventLogTestHelper.writeEventsToRollingWriter(fs, appId, dir, sparkConf,
hadoopConf, eventsFiles*)`. How about moving it to `EventLogTestHelper` as
`writeLinesToRollingWriter`? The events version could then call it with
`eventsFiles.map(_.map(convertEvent))`, so the roll/stop/list logic is kept in
one place.
##########
core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala:
##########
@@ -111,21 +120,46 @@ class EventLogFileCompactor(
/**
* Loads all available EventFilterBuilders in classloader via ServiceLoader,
and initializes
* them via replaying events in given files.
+ *
+ * Returns None if replay does not fully read the given files: a line over
the line-length
+ * limit is skipped, or replay stops in the middle of a file on an error.
The builders have not
+ * seen the missed events (e.g. the start of a live job), so filters created
from them could
+ * reject events that are still needed, and those events would be lost once
the original files
+ * are removed.
*/
- private def initializeBuilders(fs: FileSystem, files: Seq[Path]):
Seq[EventFilterBuilder] = {
+ private def initializeBuilders(
+ fs: FileSystem,
+ files: Seq[Path]): Option[Seq[EventFilterBuilder]] = {
val bus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(sparkConf))
val builders = ServiceLoader.load(classOf[EventFilterBuilder],
Utils.getContextOrSparkClassLoader).asScala.toSeq
builders.foreach(bus.addListener)
- files.foreach { log =>
- Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
+ // Stop at the first file that is not fully replayed, as compaction will
not proceed. The
+ // check happens between files, so that file itself is still read to the
end.
+ val replayedAll = files.forall { log =>
+ val completed =
Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
bus.replay(in, log.getName)
}
+ if (bus.numSkippedLines > 0) {
+ logWarning(log"Skipping compaction of event log files in " +
+ log"${MDC(LogKeys.PATH, log.getParent)}: ${MDC(LogKeys.FILE_NAME,
log.getName)} has " +
+ log"lines longer than ${MDC(LogKeys.CONFIG,
History.EVENT_LOG_MAX_LINE_LENGTH.key)} " +
+ log"(${MDC(LogKeys.MAX_SIZE,
ReplayListenerBus.maxLineLength(sparkConf))} bytes). " +
+ log"Compaction stays skipped until the limit is increased.")
+ false
+ } else if (!completed) {
+ logWarning(log"Skipping compaction of event log files in " +
+ log"${MDC(LogKeys.PATH, log.getParent)}: failed to replay " +
Review Comment:
nit. This case also lasts until the files change: the file that failed stays
in the range, so every later attempt fails at the same line, and no config can
fix it. The warning for over-long lines says it persists, but this one doesn't,
and the new paragraph in `docs/monitoring.md` mentions persistence only for
over-long lines. Could we say so here too, for example "Compaction of this
application stays skipped."?
##########
core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala:
##########
@@ -111,21 +120,46 @@ class EventLogFileCompactor(
/**
* Loads all available EventFilterBuilders in classloader via ServiceLoader,
and initializes
* them via replaying events in given files.
+ *
+ * Returns None if replay does not fully read the given files: a line over
the line-length
+ * limit is skipped, or replay stops in the middle of a file on an error.
The builders have not
+ * seen the missed events (e.g. the start of a live job), so filters created
from them could
+ * reject events that are still needed, and those events would be lost once
the original files
+ * are removed.
*/
- private def initializeBuilders(fs: FileSystem, files: Seq[Path]):
Seq[EventFilterBuilder] = {
+ private def initializeBuilders(
+ fs: FileSystem,
+ files: Seq[Path]): Option[Seq[EventFilterBuilder]] = {
val bus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(sparkConf))
val builders = ServiceLoader.load(classOf[EventFilterBuilder],
Utils.getContextOrSparkClassLoader).asScala.toSeq
builders.foreach(bus.addListener)
- files.foreach { log =>
- Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
+ // Stop at the first file that is not fully replayed, as compaction will
not proceed. The
+ // check happens between files, so that file itself is still read to the
end.
+ val replayedAll = files.forall { log =>
+ val completed =
Utils.tryWithResource(EventLogFileReader.openEventLog(log, fs)) { in =>
bus.replay(in, log.getName)
}
+ if (bus.numSkippedLines > 0) {
+ logWarning(log"Skipping compaction of event log files in " +
+ log"${MDC(LogKeys.PATH, log.getParent)}: ${MDC(LogKeys.FILE_NAME,
log.getName)} has " +
+ log"lines longer than ${MDC(LogKeys.CONFIG,
History.EVENT_LOG_MAX_LINE_LENGTH.key)} " +
+ log"(${MDC(LogKeys.MAX_SIZE,
ReplayListenerBus.maxLineLength(sparkConf))} bytes). " +
+ log"Compaction stays skipped until the limit is increased.")
Review Comment:
nit. "until the limit is increased" is not always true.
`FsHistoryProvider.compact` calls the compactor only when
`lastEvaluatedForCompaction < lastIndex`, and that is updated after this
attempt. After the History Server restarts with a higher limit, compaction runs
again only when the application rolls a new file. For an application that has
finished, or no longer rolls files, the files are not compacted when the
listing is kept in `spark.history.store.path`. How about "is skipped until the
limit is increased and a new event log file is rolled"? The same wording is in
`docs/monitoring.md` (L134 and L459) and in `History.scala`.
--
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]
