[SPARK-60109][UI][3.5] 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,15 +17,14 @@
 
 package org.apache.spark.deploy.history
 
-import java.io.{File, FileNotFoundException, IOException}
+import java.io.{ByteArrayInputStream, File, FileNotFoundException, InputStream, IOException, SequenceInputStream}
 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.JavaConverters._
 import scala.collection.mutable
-import scala.io.{Codec, Source}
 import scala.util.control.NonFatal
 import scala.xml.Node
 
@@ -832,21 +831,11 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
         val target = lastFile.getLen - reparseChunkSize
         if (target > 0) {
           logInfo(s"Looking for end event; skipping $target bytes from $logPath...")
-          var skipped = 0L
-          while (skipped < target) {
-            skipped += in.skip(target - skipped)
-          }
-        }
-
-        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()
         }
+        val (tail, startOffset) = skipToLineStart(in, target)
 
-        bus.replay(source, lastFile.getPath.toString, !appCompleted, eventsFilter)
+        bus.replayFromOffset(tail, lastFile.getPath.toString, startOffset, !appCompleted,
+          eventsFilter)
       }
     }
 
@@ -1459,6 +1448,61 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
 
 private[spark] object FsHistoryProvider {
 
+  /**
+   * Returns a stream that starts at the first line beginning at or after byte `target` of
+   * `in`, together with that line's byte offset. If `in` ends first, the returned stream is at
+   * EOF and the offset is the length of `in`. A non-positive `target` returns `in` unchanged
+   * with offset 0. Otherwise `in` is consumed and must not be read afterwards: bytes read past
+   * the line boundary are prepended to the returned stream, not pushed back into `in`.
+   *
+   * Since the skip may land anywhere, including inside a multi-byte UTF-8 character, the line
+   * boundary is found on raw bytes before any decoding: '\n' never occurs inside a multi-byte
+   * UTF-8 sequence.
+   */
+  private[history] def skipToLineStart(in: InputStream, target: Long): (InputStream, Long) = {
+    if (target <= 0) {
+      (in, 0L)
+    } else {
+      // Stop one byte early, so that a line starting exactly at `target` is kept.
+      var position = 0L
+      var eof = false
+      while (!eof && position < target - 1) {
+        val skipped = in.skip(target - 1 - position)
+        if (skipped > 0) {
+          position += skipped
+        } else if (in.read() == -1) {
+          // skip() may return 0 at EOF, e.g. when `target` comes from a compressed length that
+          // exceeds the decompressed one.
+          eof = true
+        } else {
+          position += 1
+        }
+      }
+
+      var result: InputStream = in
+      val buffer = new Array[Byte](8192)
+      var found = false
+      while (!eof && !found) {
+        val n = in.read(buffer)
+        if (n == -1) {
+          eof = true
+        } else {
+          var i = 0
+          while (i < n && buffer(i) != '\n') {
+            i += 1
+          }
+          position += math.min(i + 1, n)
+          if (i < n) {
+            found = true
+            val rest = new ByteArrayInputStream(buffer, i + 1, n - i - 1)
+            result = new SequenceInputStream(rest, in)
+          }
+        }
+      }
+      (result, position)
+    }
+  }
+
   private val APPL_START_EVENT_PREFIX = "{\"Event\":\"SparkListenerApplicationStart\""
 
   private val APPL_END_EVENT_PREFIX = "{\"Event\":\"SparkListenerApplicationEnd\""

core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala

@@ -71,8 +71,23 @@ private[spark] class ReplayListenerBus(
       sourceName: String,
       maybeTruncated: Boolean = false,
       eventsFilter: ReplayEventsFilter = SELECT_ALL_FILTER): Boolean = {
-    val lines = boundedLines(logData, sourceName)
-    replayEntries(lines, sourceName, maybeTruncated, eventsFilter)
+    replayFromOffset(logData, sourceName, 0L, maybeTruncated, eventsFilter)
+  }
+
+  /**
+   * Variant of [[replay()]] for a stream positioned at the start of a line that begins at
+   * `startOffset` bytes into the (uncompressed) source. A positive `startOffset` is added to
+   * the diagnostics so that the reported line numbers, which count from the start of the
+   * stream, can be located in the source.
+   */
+  private[spark] def replayFromOffset(
+      logData: InputStream,
+      sourceName: String,
+      startOffset: Long,
+      maybeTruncated: Boolean,
+      eventsFilter: ReplayEventsFilter): Boolean = {
+    val lines = boundedLines(logData, sourceName, startOffset)
+    replayEntries(lines, sourceName, startOffset, maybeTruncated, eventsFilter)
   }
 
   /**
@@ -83,7 +98,8 @@ private[spark] class ReplayListenerBus(
    */
   private def boundedLines(
       logData: InputStream,
-      sourceName: String): Iterator[(String, Int)] = {
+      sourceName: String,
+      startOffset: Long): Iterator[(String, Int)] = {
     // Fail on malformed input like Source.getLines() does instead of replacing it.
     val decoder = StandardCharsets.UTF_8.newDecoder()
       .onMalformedInput(CodingErrorAction.REPORT)
@@ -145,10 +161,11 @@ private[spark] class ReplayListenerBus(
             if (!warned) {
               logWarning("Skipped event log lines longer than " +
                 s"$effectiveMaxLineLength bytes in " +
-                s"$sourceName; first skipped line: ${index + 1}")
+                s"$sourceName; first skipped line: ${index + 1}" + lineOrigin(startOffset))
               warned = true
             }
-            logDebug(s"Skipped event log line ${index + 1} in $sourceName")
+            logDebug(s"Skipped event log line ${index + 1} in $sourceName" +
+              lineOrigin(startOffset))
             fetchLine()
           } else {
             (maybeLine.get, index)
@@ -167,12 +184,13 @@ private[spark] class ReplayListenerBus(
       sourceName: String,
       maybeTruncated: Boolean,
       eventsFilter: ReplayEventsFilter): Boolean = {
-    replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter)
+    replayEntries(lines.zipWithIndex, sourceName, 0L, maybeTruncated, eventsFilter)
   }
 
   private def replayEntries(
       lines: Iterator[(String, Int)],
       sourceName: String,
+      startOffset: Long,
       maybeTruncated: Boolean,
       eventsFilter: ReplayEventsFilter): Boolean = {
     var currentLine: String = null
@@ -216,7 +234,8 @@ private[spark] class ReplayListenerBus(
               throw jpe
             } else {
               logWarning(s"Got JsonParseException from log file $sourceName" +
-                s" at line $lineNumber, the file might not have finished writing cleanly.")
+                s" at line $lineNumber" + lineOrigin(startOffset) +
+                ", the file might not have finished writing cleanly.")
             }
         }
       }
@@ -228,20 +247,25 @@ private[spark] class ReplayListenerBus(
       case _: EOFException if maybeTruncated => false
       case jpe: JsonProcessingException =>
         logError(s"Exception parsing Spark event log: $sourceName " +
-          s"at line $lineNumber")
+          s"at line $lineNumber" + lineOrigin(startOffset))
         throw jpe
       case ioe: IOException =>
         throw ioe
       case e: Exception =>
         logError(s"Exception parsing Spark event log: $sourceName", e)
         val prefix = Option(currentLine).map(_.take(1024)).orNull
         val length = Option(currentLine).fold(0)(_.length)
-        logError(s"Malformed line #$lineNumber: $prefix " +
+        logError(s"Malformed line #$lineNumber" + lineOrigin(startOffset) + s": $prefix " +
           s"(line length: $length characters; showing at most 1024)\n")
         false
     }
   }
 
+  /** Describes where line numbers start when the replay does not start at the source start. */
+  private def lineOrigin(startOffset: Long): String = {
+    if (startOffset > 0) s" (lines counted from uncompressed byte offset $startOffset)" else ""
+  }
+
   override protected def isIgnorableException(e: Throwable): Boolean = {
     e.isInstanceOf[HaltReplayException]
   }