[SPARK-60109][UI] Apply the replay line-length limit and UTF-8-safe line alignment to the History Server end-event reparse

core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala

@@ -17,14 +17,13 @@
 
 package org.apache.spark.deploy.history
 
-import java.io.{File, FileNotFoundException, IOException}
+import java.io.{File, FileNotFoundException, InputStream, IOException}
 import java.lang.{Long => JLong}
 import java.util.{Date, NoSuchElementException, ServiceLoader}
 import java.util.concurrent.{ConcurrentHashMap, ExecutorService, TimeUnit}
 import java.util.zip.ZipOutputStream
 
 import scala.collection.mutable
-import scala.io.{Codec, Source}
 import scala.jdk.CollectionConverters._
 import scala.util.control.NonFatal
 import scala.xml.Node
@@ -1107,24 +1106,23 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
       Utils.tryWithResource(EventLogFileReader.openEventLog(lastFile.getPath,
           fsForPath(lastFile.getPath))) { in =>
         val target = lastFile.getLen - reparseChunkSize
-        if (target > 0) {
+        val sourceName = if (target > 0) {
           logInfo(log"Looking for end event; skipping ${MDC(NUM_BYTES, target)} bytes" +
             log" from ${MDC(PATH, logPath)}...")
           var skipped = 0L
           while (skipped < target) {
             skipped += in.skip(target - skipped)
           }
+          // Because skipping may leave the stream in the middle of a line, or even in the
+          // middle of a multi-byte UTF-8 character, discard the rest of that line before any
+          // decoding. Line numbers reported by the replay are then relative to this offset.
+          val offset = target + skipPartialLine(in)
+          s"${lastFile.getPath} (lines counted from uncompressed byte offset $offset)"
+        } else {
+          lastFile.getPath.toString
         }
 
-        val source = Source.fromInputStream(in)(Codec.UTF8).getLines()
-
-        // Because skipping may leave the stream in the middle of a line, read the next line
-        // before replaying.
-        if (target > 0) {
-          source.next()
-        }
-
-        bus.replay(source, lastFile.getPath.toString, !appCompleted, eventsFilter)
+        bus.replay(in, sourceName, !appCompleted, eventsFilter)
       }
     }
 
@@ -1772,6 +1770,22 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
 
 private[spark] object FsHistoryProvider {
 
+  /**
+   * Discards bytes up to and including the next '\n', returning the number of bytes discarded.
+   * This works on raw bytes so that it is safe to call at an arbitrary byte offset: '\n' never
+   * occurs inside a multi-byte UTF-8 sequence. The bytes are read one at a time so that the
+   * stream is left exactly at the start of the next line.
+   */
+  private[history] def skipPartialLine(in: InputStream): Long = {
+    var discarded = 0L
+    var b = in.read()
+    while (b != -1 && b != '\n') {
+      discarded += 1
+      b = in.read()
+    }
+    if (b == '\n') discarded + 1 else discarded
+  }
+
   private val APPL_START_EVENT_PREFIX = "{\"Event\":\"SparkListenerApplicationStart\""
 
   private val APPL_END_EVENT_PREFIX = "{\"Event\":\"SparkListenerApplicationEnd\""

core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala

@@ -1248,6 +1248,61 @@ abstract class FsHistoryProviderSuite extends SparkFunSuite with Matchers with P
     }
   }
 
+  test("end event reparse when the skip offset lands inside a multi-byte character") {
+    val log = newLogFile("end-event-utf8", None, inProgress = false)
+    val bytes = writeRawLog(log, endEventTestLines("end-event-utf8",
+      jobStartLine(1, 0x4e2d.toChar.toString * 200)))
+    // The job description is written as raw UTF-8: skip to the second byte of a character.
+    val leadByte = bytes.indexWhere(b => (b & 0xF0) == 0xE0)
+    assert(leadByte > 0)
+    val conf = createTestConf().set(END_EVENT_REPARSE_CHUNK_SIZE, (bytes.length - leadByte - 1L))
+    updateAndCheck(new FsHistoryProvider(conf)) { list =>
+      assert(list.size === 1)
+      assert(list(0).attempts.head.completed)
+    }
+  }
+
+  test("end event reparse applies the line length limit and reports relative line numbers") {
+    val log = newLogFile("end-event-limit", None, inProgress = false)
+    val tail = Seq(jobStartLine(2, "x" * 2048), jobStartLine(3, "y"))
+    val lines = endEventTestLines("end-event-limit", tail: _*)
+    val bytes = writeRawLog(log, lines)
+    // Skip into the middle of the line before the tail. The next line starts at `offset` and
+    // the over-long line is the first line after it.
+    val offset = lines.takeWhile(_ != tail.head).map(_.length + 1).sum
+    val conf = createTestConf()
+      .set(END_EVENT_REPARSE_CHUNK_SIZE, bytes.length - offset + 10L)
+      .set(EVENT_LOG_MAX_LINE_LENGTH.key, "1k")
+    val appender = new LogAppender("end event reparse")
+    withLogAppender(appender) {
+      updateAndCheck(new FsHistoryProvider(conf)) { list =>
+        assert(list.size === 1)
+        assert(list(0).attempts.head.completed)
+      }
+    }
+    val sourceName = s"${log.getName} (lines counted from uncompressed byte offset $offset)"
+    val messages = appender.loggingEvents.map(_.getMessage.getFormattedMessage)
+    assert(messages.exists(m => m.contains("Skipped event log lines longer than 1024 bytes") &&
+      m.contains(s"$sourceName; first skipped line: 1")), messages)
+  }
+
+  test("end event reparse labels JSON errors with the skip offset") {
+    val log = newLogFile("end-event-malformed", None, inProgress = false)
+    val malformed = "{\"Event\":\"SparkListenerApplicationEnd\",\"Timestamp\":"
+    val tail = Seq(jobStartLine(2, "x"), malformed, jobStartLine(3, "y"))
+    val lines = endEventTestLines("end-event-malformed", tail: _*)
+    val bytes = writeRawLog(log, lines)
+    val offset = lines.takeWhile(_ != tail.head).map(_.length + 1).sum
+    val conf = createTestConf().set(END_EVENT_REPARSE_CHUNK_SIZE, bytes.length - offset + 10L)
+    val appender = new LogAppender("end event reparse")
+    withLogAppender(appender) {
+      new FsHistoryProvider(conf).checkForLogs()
+    }
+    val sourceName = s"${log.getName} (lines counted from uncompressed byte offset $offset)"
+    val messages = appender.loggingEvents.map(_.getMessage.getFormattedMessage)
+    assert(messages.exists(_.contains(s"$sourceName at line 2")), messages)
+  }
+
   test("parse event logs with optimizations off") {
     val conf = createTestConf()
       .set(END_EVENT_REPARSE_CHUNK_SIZE, 0L)
@@ -2156,6 +2211,39 @@ abstract class FsHistoryProviderSuite extends SparkFunSuite with Matchers with P
     }
   }
 
+  /** Writes the given lines as an uncompressed event log and returns the written bytes. */
+  private def writeRawLog(file: File, lines: Seq[String]): Array[Byte] = {
+    val bytes = lines.map(_ + "\n").mkString.getBytes(StandardCharsets.UTF_8)
+    Utils.tryWithResource(new FileOutputStream(file))(_.write(bytes))
+    bytes
+  }
+
+  private def jobStartLine(jobId: Int, description: String): String = {
+    val properties = new java.util.Properties()
+    properties.setProperty("spark.job.description", description)
+    JsonProtocol.sparkEventToJsonString(SparkListenerJobStart(jobId, 1L, Nil, properties))
+  }
+
+  /**
+   * Event log lines with enough information for a listing entry, then a padding line, the given
+   * tail lines and the application end event.
+   */
+  private def endEventTestLines(appId: String, tail: String*): Seq[String] = {
+    Seq(
+      SparkListenerLogStart(SPARK_VERSION),
+      SparkListenerApplicationStart(appId, Some(appId), 1L, "test", 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))
+    ).map(JsonProtocol.sparkEventToJsonString) ++
+      Seq(jobStartLine(1, "padding")) ++ tail ++
+      Seq(JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(5L)))
+  }
+
   private def createEmptyFile(file: File) = {
     new FileOutputStream(file).close()
   }