HyukjinKwon commented on code in PR #58700:
URL: https://github.com/apache/spark/pull/58700#discussion_r3985271016
##########
core/src/main/scala/org/apache/spark/internal/config/History.scala:
##########
@@ -165,6 +165,19 @@ private[spark] object History {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("1m")
+ 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. " +
+ "Introduced in 4.3.0; also available in 3.5.10, 4.0.5, 4.1.4 and
4.2.1; and in " +
+ "all versions after 4.3.0.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("512m")
Review Comment:
This is declared as a byte size (`bytesConf`, `512m`), but the cap is
enforced as a UTF-16 character count downstream — `sb.length() < maxLineLength`
in `ReplayListenerBus.boundedLines`. So for multi-byte UTF-8 the effective byte
threshold is larger than configured, and a retained line can hold up to ~2x the
configured number of bytes in heap (Java `char` is 2 bytes).
The class parameter doc already says "characters"; worth saying the same in
this entry's `.doc(...)` and in the `docs/monitoring.md` row so the documented
contract matches enforcement.
##########
core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala:
##########
@@ -56,10 +66,79 @@ private[spark] class ReplayListenerBus extends
SparkListenerBus with Logging {
sourceName: String,
maybeTruncated: Boolean = false,
eventsFilter: ReplayEventsFilter = SELECT_ALL_FILTER): Boolean = {
- val lines = Source.fromInputStream(logData)(Codec.UTF8).getLines()
+ val lines = boundedLines(logData, sourceName)
replay(lines, sourceName, maybeTruncated, eventsFilter)
}
+ /**
+ * Reads '\n'-terminated lines like Source.getLines(), but never
materializes more than
+ * [[maxLineLength]] characters of a single line. An over-long line is
drained and skipped
Review Comment:
A `[[...]]` wiki-link only resolves to documented members, and
`maxLineLength` is a non-`val` constructor parameter, which Scaladoc doesn't
document — so this renders as literal text rather than a link.
```suggestion
* `maxLineLength` characters of a single line. An over-long line is
drained and skipped
```
--
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]
