You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sa...@apache.org on 2026/10/20 00:01:57 UTC
(spark) branch master updated: [SPARK-59747][CORE] Don't retry compaction of rolled event log files with malformed input
This is an automated email from the ASF dual-hosted git repository.
sarutak pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new d9a35e971b22 [SPARK-59747][CORE] Don't retry compaction of rolled event log files with malformed input
d9a35e971b22 is described below
commit d9a35e971b22f2c663fdf4738442c53d1a33d020
Author: Yiping Deng <yi...@gmail.com>
AuthorDate: Tue Oct 20 09:01:46 2026 +0900
[SPARK-59747][CORE] Don't retry compaction of rolled event log files with malformed input
### What changes were proposed in this pull request?
In `FsHistoryProvider.compact()`, catch `CharacterCodingException` thrown while compacting rolled event log files, log it at WARN, and still record `lastEvaluatedForCompaction` for the current index.
`EventLogFileCompactor.initializeBuilders` also logs a WARN naming the malformed event log file before rethrowing the decoding exception.
### Why are the changes needed?
When the rolled event log files that are eligible for compaction contain bytes that are not valid UTF-8, `EventLogFileCompactor` fails with a `MalformedInputException` while replaying them to initialize the event filter builders. `compact()` only records `lastEvaluatedForCompaction` after compaction succeeds, and the exception reaches the generic handler, which logs it at ERROR.
The listing replay stops early, so it never reaches the malformed bytes and the application keeps being listed. Each time the last (non-compacted) event log file grows, the log is reprocessed and compaction of the same unchanged rolled files is tried again. For a long-running application this logs a full stack trace at ERROR on every scan until a new event log file is rolled. Rolled files are no longer modified, so the retry can never succeed.
With this change, compaction for that index is skipped, and it is only tried again once a new event log file is rolled. Other failures are handled as before.
### Does this PR introduce _any_ user-facing change?
Yes, but only in History Server logs. When compaction fails because the event log files contain malformed input, the History Server logs a WARN once per rolled event log index instead of an ERROR on every scan. An additional WARN identifies the path of the malformed event log file, alongside the WARN for the application directory. The application listing and replay are unchanged, and the event log files are left uncompacted as before.
### How was this patch tested?
Added a new test to `FsHistoryProviderSuite`. It writes a rolled event log file with a malformed UTF-8 sequence after the events used for listing, then checks that the application is still listed, no compact file is created, `lastEvaluatedForCompaction` is recorded, and a WARN names the malformed file. The test fails without this change (`None did not equal Some(2)`, with `MalformedInputException` logged at ERROR) and passes with it.
```
build/mvn -pl core -am test -Dtest=none -DwildcardSuites=org.apache.spark.deploy.history
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #58996 from DengYiping/shs-compaction-malformed-input.
Authored-by: Yiping Deng <yi...@gmail.com>
Signed-off-by: Kousuke Saruta <sa...@apache.org>
---
.../deploy/history/EventLogFileCompactor.scala | 9 ++-
.../spark/deploy/history/FsHistoryProvider.scala | 14 ++++-
.../deploy/history/FsHistoryProviderSuite.scala | 65 +++++++++++++++++++++-
3 files changed, 84 insertions(+), 4 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala
index 2e54ae4db9b3..cdec1453b4de 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala
@@ -19,6 +19,7 @@ package org.apache.spark.deploy.history
import java.io.IOException
import java.net.URI
+import java.nio.charset.CharacterCodingException
import java.util.ServiceLoader
import scala.jdk.CollectionConverters._
@@ -140,7 +141,13 @@ class EventLogFileCompactor(
// 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)
+ try {
+ bus.replay(in, log.getName)
+ } catch {
+ case e: CharacterCodingException =>
+ logWarning(log"Event log file ${MDC(LogKeys.PATH, log)} contains malformed input")
+ throw e
+ }
}
if (bus.numSkippedLines > 0) {
logWarning(log"Skipping compaction of event log files in " +
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
index 7f984d5a0578..354211661f38 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
@@ -19,6 +19,7 @@ package org.apache.spark.deploy.history
import java.io.{ByteArrayInputStream, File, FileNotFoundException, InputStream, IOException, SequenceInputStream}
import java.lang.{Long => JLong}
+import java.nio.charset.CharacterCodingException
import java.util.{Date, NoSuchElementException, ServiceLoader}
import java.util.concurrent.{ConcurrentHashMap, ExecutorService, TimeUnit}
import java.util.zip.ZipOutputStream
@@ -1194,7 +1195,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 =>
+ // rewrite() only runs after initializeBuilders has replayed every file to compact
+ // without error, so no files have been written or deleted at this point.
+ // 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 " +
+ log"log files contain malformed input", e)
+ }
listing.write(info.copy(lastEvaluatedForCompaction = Some(lastIndex)))
}
} catch {
diff --git a/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala b/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala
index 304920813325..b46bba5c7ba6 100644
--- a/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala
@@ -19,7 +19,7 @@ package org.apache.spark.deploy.history
import java.io._
import java.nio.charset.StandardCharsets
-import java.nio.file.Files
+import java.nio.file.{Files, StandardOpenOption}
import java.util.{Date, Locale}
import java.util.concurrent.TimeUnit
import java.util.zip.{ZipInputStream, ZipOutputStream}
@@ -30,6 +30,7 @@ import org.apache.hadoop.fs.{FileStatus, FileSystem, FSDataInputStream, Path}
import org.apache.hadoop.hdfs.{DFSInputStream, DistributedFileSystem}
import org.apache.hadoop.ipc.{CallerContext => HadoopCallerContext}
import org.apache.hadoop.security.AccessControlException
+import org.apache.logging.log4j.Level
import org.mockito.ArgumentMatchers.{any, argThat}
import org.mockito.Mockito.{doThrow, mock, spy, verify, when}
import org.scalatest.PrivateMethodTester
@@ -40,7 +41,7 @@ import org.scalatest.matchers.should.Matchers._
import org.apache.spark.{JobExecutionStatus, SecurityManager, SPARK_VERSION, SparkConf, SparkFunSuite}
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.deploy.history.EventLogTestHelper._
-import org.apache.spark.internal.config.DRIVER_LOG_DFS_DIR
+import org.apache.spark.internal.config.{DRIVER_LOG_DFS_DIR, EVENT_LOG_COMPRESS}
import org.apache.spark.internal.config.History._
import org.apache.spark.internal.config.UI.{ADMIN_ACLS, ADMIN_ACLS_GROUPS, UI_VIEW_ACLS, UI_VIEW_ACLS_GROUPS, USER_GROUPS_MAPPING}
import org.apache.spark.io._
@@ -1775,6 +1776,66 @@ abstract class FsHistoryProviderSuite extends SparkFunSuite with Matchers with P
}
}
+ test("SPARK-59747: don't retry compaction of event log files with malformed input") {
+ withTempDir { dir =>
+ val conf = createTestConf().set(FAST_IN_PROGRESS_PARSING, true)
+ conf.set(HISTORY_LOG_DIR, dir.getAbsolutePath)
+ conf.set(EVENT_LOG_ROLLING_MAX_FILES_TO_RETAIN, 1)
+ conf.set(EVENT_LOG_COMPACTION_SCORE_THRESHOLD, 0.0d)
+ // Keep the event log files uncompressed so that the bytes appended below are decoded as-is.
+ conf.set(EVENT_LOG_COMPRESS, false)
+ val hadoopConf = SparkHadoopUtil.newConfiguration(conf)
+ val fs = new Path(dir.getAbsolutePath).getFileSystem(hadoopConf)
+
+ val provider = new FsHistoryProvider(conf)
+
+ val writer = new RollingEventLogFilesWriter("app", None, dir.toURI, conf, hadoopConf)
+ writer.start()
+
+ // Keep the writer running so fast listing replay halts after the environment update.
+ // Pad the first file so the malformed bytes are not decoded by listing's read-ahead,
+ // but are decoded when compaction replays the file in full.
+ writeEventsToRollingWriter(writer, Seq(
+ SparkListenerApplicationStart("app", Some("app"), 0, "user", None),
+ SparkListenerEnvironmentUpdate(Map(
+ "Spark Properties" -> Seq.empty,
+ "Hadoop Properties" -> Seq.empty,
+ "JVM Information" -> Seq.empty,
+ "System Properties" -> Seq.empty,
+ "Metrics Properties" -> Seq.empty,
+ "Classpath Entries" -> Seq.empty))) ++
+ (1 to 1000).map(SparkListenerJobStart(_, 0, Seq.empty)), rollFile = true)
+
+ val firstFile = EventLogFileReader(fs, new Path(writer.logPath)).get.listEventLogFiles.head
+ Files.write(new File(firstFile.getPath.toUri).toPath,
+ Array[Byte](0xC3.toByte, 0x28.toByte, '\n'.toByte), StandardOpenOption.APPEND)
+
+ writeEventsToRollingWriter(writer, Seq(SparkListenerJobEnd(1, 1, JobSucceeded)),
+ rollFile = false)
+
+ val logAppender = new LogAppender("malformed event log compaction")
+ withLogAppender(logAppender, loggerNames = Seq(classOf[EventLogFileCompactor].getName)) {
+ updateAndCheck(provider) { list =>
+ assert(list.map(_.id) === Seq("app"))
+ val logFiles = EventLogFileReader(fs, new Path(writer.logPath)).get.listEventLogFiles
+ assert(!logFiles.exists(file => EventLogFileWriter.isCompacted(file.getPath)))
+ assert(logFiles.map { file =>
+ RollingEventLogFilesWriter.getEventLogFileIndex(file.getPath.getName)
+ } === Seq(1, 2))
+ val info = provider.listing.read(classOf[LogInfo], writer.logPath)
+ assert(info.lastEvaluatedForCompaction === Some(2))
+ }
+ }
+ assert(logAppender.loggingEvents.exists { event =>
+ event.getLevel == Level.WARN && event.getMessage.getFormattedMessage ==
+ s"Event log file ${firstFile.getPath} contains malformed input"
+ })
+
+ writer.stop()
+ provider.stop()
+ }
+ }
+
test("SPARK-33146: don't let one bad rolling log folder prevent loading other applications") {
withTempDir { dir =>
val conf = createTestConf(true)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org