This is an automated email from the ASF dual-hosted git repository.
sarutak pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new cda6c10610f0 [SPARK-59747][4.1][CORE] Don't retry compaction of rolled
event log files with malformed input
cda6c10610f0 is described below
commit cda6c10610f0de14c0fdf1221d4ce68cf2a7fe51
Author: Yiping Deng <[email protected]>
AuthorDate: Thu Oct 29 05:24:09 2026 +0900
[SPARK-59747][4.1][CORE] Don't retry compaction of rolled event log files
with malformed input
### What changes were proposed in this pull request?
Backport #58996 to `branch-4.1`.
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.
The conflict resolution retains this branch's existing single
`fileCompactor`.
### 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.
Passed locally with Java 21 on the `branch-4.1` backport:
```sh
build/sbt 'core/Compile/scalastyle' 'core/Test/scalastyle' \
'core/testOnly
org.apache.spark.deploy.history.RocksDBBackendFsHistoryProviderSuite
org.apache.spark.deploy.history.RocksDBBackendWithProtobufSerializerSuite -- -z
"SPARK-59747"' \
'core/testOnly org.apache.spark.deploy.history.EventLogFileCompactorSuite'
```
Both malformed-input regression variants passed (2 tests), including the
WARN naming the malformed file. All 12 compactor tests passed, including the
SPARK-60110 incomplete-replay coverage. Both style checks passed with zero
errors/warnings; `git diff --check` also passed. LevelDB-backed variants were
not run on Apple Silicon.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5) for the original patch; OpenAI
Codex (GPT-5) for the backport.
Closes #59619 from DengYiping/ydeng/spark-59747-branch-4.1.
Authored-by: Yiping Deng <[email protected]>
Signed-off-by: Kousuke Saruta <[email protected]>
---
.../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 df7ee23c55e1..5f1bbaa67a7a 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
@@ -946,7 +947,18 @@ private[history] class FsHistoryProvider(conf: SparkConf,
clock: Clock)
if (info.lastEvaluatedForCompaction.isEmpty ||
info.lastEvaluatedForCompaction.get < lastIndex) {
// haven't tried compaction for this index, do compaction
- fileCompactor.compact(reader.listEventLogFiles)
+ try {
+ fileCompactor.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 8ba6aef7c905..afeda85e9bde 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._
@@ -1722,6 +1723,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: [email protected]
For additional commands, e-mail: [email protected]
