viirya commented on code in PR #59328:
URL: https://github.com/apache/spark/pull/59328#discussion_r4265555073
##########
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:
Good point. Changed it to "until the limit is increased and a new event log
file is rolled" in the warning, in both places in `docs/monitoring.md`, and in
`History.scala`.
##########
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:
Done. The warning now ends with "Compaction of this application stays
skipped.", and the paragraph in `docs/monitoring.md` now says that compaction
stays skipped in both cases, and that for an over-long line it resumes once the
limit is increased and a new event log file is rolled.
##########
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:
Done, moved it to `EventLogTestHelper.writeLinesToRollingWriter`, and
`writeEventsToRollingWriter` now calls it with
`eventsFiles.map(_.map(convertEvent))`.
--
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]
