viirya commented on code in PR #58996:
URL: https://github.com/apache/spark/pull/58996#discussion_r4265677626
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1205,7 +1206,18 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
info.lastEvaluatedForCompaction.get < lastIndex) {
// haven't tried compaction for this index, do compaction
val compactor = fileCompactors(logDirForPath(reader.rootPath))
- compactor.compact(reader.listEventLogFiles)
+ try {
+ compactor.compact(reader.listEventLogFiles)
+ } catch {
+ case e: CharacterCodingException =>
+ // This relies on initializeBuilders decoding every input
file in full before
+ // rewrite() creates a compact file, so no files have been
written or deleted.
+ // Only rolled event log files are compacted and they are no
longer modified,
+ // so retrying would fail the same way. Record this index as
evaluated to
+ // avoid replaying the same files on every scan.
+ logWarning(log"Skipping compaction for ${MDC(PATH,
rootPath)} as the event " +
Review Comment:
The exception message here is just `Input length = 1`, and
`ReplayListenerBus` rethrows `IOException` without logging the source, so this
WARN only identifies the application directory, not which rolled file is
corrupted. Since the same file also breaks loading the application UI
(`parseAppEventLogs` doesn't catch this either), it would help operators to
know which file to look at. Is there a simple way to include it here, or would
it be easier to log it where the file is known?
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1205,7 +1206,18 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
info.lastEvaluatedForCompaction.get < lastIndex) {
// haven't tried compaction for this index, do compaction
val compactor = fileCompactors(logDirForPath(reader.rootPath))
- compactor.compact(reader.listEventLogFiles)
+ try {
+ compactor.compact(reader.listEventLogFiles)
+ } catch {
+ case e: CharacterCodingException =>
+ // This relies on initializeBuilders decoding every input
file in full before
Review Comment:
With #59328, `initializeBuilders` stops at the first file that is not fully
replayed, so "decoding every input file in full" will no longer hold literally,
even though the safety argument still does. Could we phrase it in terms of what
this actually relies on? Something like:
```scala
// rewrite() only runs after initializeBuilders has replayed every file to
compact
// without error, so no files have been written or deleted at this point.
```
--
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]
