viirya commented on code in PR #59074:
URL: https://github.com/apache/spark/pull/59074#discussion_r4126668723
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -67,22 +69,26 @@ private[spark] class ReplayListenerBus(
maybeTruncated: Boolean = false,
eventsFilter: ReplayEventsFilter = SELECT_ALL_FILTER): Boolean = {
val lines = boundedLines(logData, sourceName)
- replay(lines, sourceName, maybeTruncated, eventsFilter)
+ replayEntries(lines, sourceName, maybeTruncated, eventsFilter)
}
/**
- * Reads '\n'-terminated lines like Source.getLines(), but never
materializes more than
- * `maxLineLength` bytes of a single line. An over-long line is drained and
skipped
- * with a warning instead of being turned into a String.
+ * Reads '\n'-terminated lines and retains their original zero-based
indices. The limit
+ * measures UTF-8 content bytes, excluding the line ending, rather than JVM
heap usage.
+ * The character buffer is bounded by the limit plus one possible trailing
CR. An over-long
+ * line is drained and skipped with a warning instead of being turned into a
String.
*/
- private def boundedLines(logData: InputStream, sourceName: String):
Iterator[String] = {
+ private def boundedLines(
+ logData: InputStream,
+ sourceName: String): Iterator[(String, Int)] = {
// Fail on malformed input like Source.getLines() does instead of
replacing it.
val decoder = StandardCharsets.UTF_8.newDecoder()
.onMalformedInput(CodingErrorAction.REPORT)
.onUnmappableCharacter(CodingErrorAction.REPORT)
val reader = new BufferedReader(new InputStreamReader(logData, decoder))
Review Comment:
We will evaluate raw-byte splitting in the reader follow-up, preserving
strict validation of skipped content and covering tail alignment and incomplete
UTF-8 at EOF. I have not independently reproduced the benchmark numbers.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +110,29 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
- val sb = new java.lang.StringBuilder()
- var overLong = false
+ @tailrec private def fetchLine(): (String, Int) = {
+ val buffer = new BoundedLineBuffer(effectiveMaxLineLength)
var c = reader.read()
if (c == -1) {
null
} else {
+ val index = lineIndex
Review Comment:
I would keep the current iterator in this fix to limit the backport diff. We
can revisit the iterator structure with the reader follow-up.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -104,35 +110,29 @@ private[spark] class ReplayListenerBus(
line
}
- @tailrec private def fetchLine(): String = {
- val sb = new java.lang.StringBuilder()
- var overLong = false
+ @tailrec private def fetchLine(): (String, Int) = {
+ val buffer = new BoundedLineBuffer(effectiveMaxLineLength)
Review Comment:
Agreed to evaluate buffer reuse in the reader follow-up, including a
retention cap after unusually large lines. I have not independently reproduced
the allocation or throughput measurements.
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +112,166 @@ class ReplayListenerSuite extends SparkFunSuite with
BeforeAndAfter with LocalSp
assert(eventMonster.loggedEvents(1) ===
JsonProtocol.sparkEventToJsonString(applicationEnd))
}
+ test("Replay line limit uses UTF-8 bytes") {
Review Comment:
Added the SPARK-59804 prefix to the new test names.
##########
core/src/main/scala/org/apache/spark/internal/config/History.scala:
##########
@@ -167,16 +167,19 @@ private[spark] object History {
val EVENT_LOG_MAX_LINE_LENGTH =
ConfigBuilder("spark.history.fs.eventLog.maxLineLength")
- .doc("Maximum length of a single event log line during replay. Lines
longer than " +
- "this are skipped with a warning instead of being read into memory,
bounding the " +
- "memory replay can use when an event log is corrupt or unexpectedly
large. Setting " +
- "this to 0 or a negative value disables the limit. " +
+ .doc("Maximum UTF-8 byte length of a single event log line during
replay, excluding " +
+ "the line ending. Longer lines are skipped with a warning, bounding
the " +
+ "memory replay can use when an event log is corrupt or unexpectedly
large. Buffer " +
+ "growth, UTF-16 storage and JSON parsing can require several times
this limit in heap " +
+ "space. Values at or below 0, or above 536870912, use the maximum
supported limit of " +
Review Comment:
The maximum now lives in History, and both the config documentation and
ReplayListenerBus derive it from that constant. monitoring.md retains the
corresponding documented value.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -35,15 +35,17 @@ import org.apache.spark.util.JsonProtocol
/**
* A SparkListenerBus that can be used to replay events from serialized event
data.
*
- * @param maxLineLength Maximum number of byes (1/2 chars) of a single event
log line that will be
- * materialized during replay. Longer lines are drained,
skipped and
- * logged, bounding the memory replay can use when an
event log is
- * corrupt or unexpectedly large.
+ * @param maxLineLength Maximum UTF-8 byte length of a single event log line,
excluding its line
Review Comment:
The constructor scaladoc now describes the normalization of non-positive and
above-maximum values to the 512 MiB limit.
--
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]
