diff --git a/core/src/main/scala/org/apache/spark/internal/config/History.scala b/core/src/main/scala/org/apache/spark/internal/config/History.scala index 7353481b6582f..b75500f7e5c34 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/History.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/History.scala @@ -165,18 +165,25 @@ private[spark] object History { .bytesConf(ByteUnit.BYTE) .createWithDefaultString("1m") + val EVENT_LOG_MAX_LINE_LENGTH_LIMIT: Int = 512 * 1024 * 1024 + 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. Buffer capacity grows in steps: reducing 256m to 200m or 150m may not reduce " + + "the retained buffer; use 128m to reach a smaller capacity. Values at or below 0, " + + s"or above $EVENT_LOG_MAX_LINE_LENGTH_LIMIT, use the maximum supported limit of " + + s"$EVENT_LOG_MAX_LINE_LENGTH_LIMIT bytes (512 MiB). This cap avoids JVM array-size " + + "limits but does not guarantee sufficient heap space. " + "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") + .createWithDefaultString("256m") private[spark] val EVENT_LOG_ROLLING_MAX_FILES_TO_RETAIN = ConfigBuilder("spark.history.fs.eventLog.rolling.maxFilesToRetain") diff --git a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala b/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala index 1cd00facb7396..e2d7892428f60 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala @@ -22,7 +22,7 @@ import java.nio.charset.{CodingErrorAction, StandardCharsets} import scala.annotation.tailrec -import com.fasterxml.jackson.core.JsonParseException +import com.fasterxml.jackson.core.{JsonParseException, JsonProcessingException} import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException import org.apache.spark.SparkConf @@ -35,15 +35,21 @@ 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 + * ending. Longer lines are drained, skipped and logged, bounding the + * memory replay can use when an event log is corrupt or unexpectedly large. + * Values at or below zero or above MAX_LINE_LENGTH use MAX_LINE_LENGTH + * (512 MiB). */ private[spark] class ReplayListenerBus( maxLineLength: Int = ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH) extends SparkListenerBus with Logging { + def this() = this(ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH) + + private[scheduler] val effectiveMaxLineLength = + ReplayListenerBus.normalizeMaxLineLength(maxLineLength) + /** * Replay each event in the order maintained in the given stream. The stream is expected to * contain one JSON-encoded SparkListenerEvent per line. @@ -67,22 +73,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)) - new Iterator[String] { - private var nextLine: String = _ + new Iterator[(String, Int)] { + private var nextLine: (String, Int) = _ + private var lineIndex = 0 private var lineFetched = false private var warned = false @@ -94,7 +104,7 @@ private[spark] class ReplayListenerBus( nextLine != null } - override def next(): String = { + override def next(): (String, Int) = { if (!hasNext) { throw new NoSuchElementException("No more lines") } @@ -104,35 +114,30 @@ 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 + lineIndex += 1 while (c != -1 && c != '\n') { - if (sb.length() * 2 < maxLineLength) { - sb.append(c.toChar) - } else { - overLong = true - } + buffer.append(c.toChar) c = reader.read() } - if (overLong) { + val maybeLine = buffer.result() + if (maybeLine.isEmpty) { if (!warned) { logWarning(log"Skipped event log lines longer than " + - log"${MDC(MAX_SIZE, maxLineLength)} characters in " + - log"${MDC(FILE_NAME, sourceName)}") + log"${MDC(MAX_SIZE, effectiveMaxLineLength)} bytes in " + + log"${MDC(FILE_NAME, sourceName)}; first skipped line: ${MDC(LINE_NUM, index + 1)}") warned = true } + logDebug(s"Skipped event log line ${index + 1} in $sourceName") fetchLine() } else { - // Handle CRLF line endings like Source.getLines() does. - if (sb.length() > 0 && sb.charAt(sb.length() - 1) == '\r') { - sb.setLength(sb.length() - 1) - } - sb.toString + (maybeLine.get, index) } } } @@ -148,15 +153,21 @@ private[spark] class ReplayListenerBus( sourceName: String, maybeTruncated: Boolean, eventsFilter: ReplayEventsFilter): Boolean = { + replayEntries(lines.zipWithIndex, sourceName, maybeTruncated, eventsFilter) + } + + private def replayEntries( + lines: Iterator[(String, Int)], + sourceName: String, + maybeTruncated: Boolean, + eventsFilter: ReplayEventsFilter): Boolean = { var currentLine: String = null var lineNumber: Int = 0 val unrecognizedEvents = new scala.collection.mutable.HashSet[String] val unrecognizedProperties = new scala.collection.mutable.HashSet[String] try { - val lineEntries = lines - .zipWithIndex - .filter { case (line, _) => eventsFilter(line) } + val lineEntries = lines.filter { case (line, _) => eventsFilter(line) } while (lineEntries.hasNext) { try { @@ -202,11 +213,18 @@ private[spark] class ReplayListenerBus( // Just stop replay. false case _: EOFException if maybeTruncated => false + case jpe: JsonProcessingException => + logError(log"Exception parsing Spark event log: ${MDC(PATH, sourceName)} " + + log"at line ${MDC(LINE_NUM, lineNumber)}") + throw jpe case ioe: IOException => throw ioe case e: Exception => logError(log"Exception parsing Spark event log: ${MDC(PATH, sourceName)}", e) - logError(log"Malformed line #${MDC(LINE_NUM, lineNumber)}: ${MDC(LINE, currentLine)}\n") + val prefix = Option(currentLine).map(_.take(1024)).orNull + val length = Option(currentLine).fold(0)(_.length) + logError(log"Malformed line #${MDC(LINE_NUM, lineNumber)}: ${MDC(LINE, prefix)} " + + log"(line length: ${MDC(SIZE, length)} characters; showing at most 1024)\n") false } } @@ -225,21 +243,62 @@ private[spark] class HaltReplayException extends RuntimeException private[spark] object ReplayListenerBus { - /** - * Default per-line cap during replay: far above any legitimate event line, so replay - * memory stays bounded even for corrupt logs. Matches the default of - * spark.history.fs.eventLog.maxLineLength. - */ - val DEFAULT_MAX_LINE_LENGTH: Int = 512 * 1024 * 1024 + /** Default UTF-8 content limit shared with the history server configuration. */ + val DEFAULT_MAX_LINE_LENGTH: Int = History.EVENT_LOG_MAX_LINE_LENGTH.defaultValue.get.toInt + + // Bound StringBuilder growth so UTF-16 inflation stays below the JVM array-size limit. + val MAX_LINE_LENGTH: Int = History.EVENT_LOG_MAX_LINE_LENGTH_LIMIT - /** Resolves the replay line-length cap from configuration; <= 0 disables the cap. */ + /** Resolves the byte limit, using 512 MiB for non-positive or larger values. */ def maxLineLength(conf: SparkConf): Int = { - val configured = conf.get(History.EVENT_LOG_MAX_LINE_LENGTH) - if (configured <= 0 || configured > Int.MaxValue) Int.MaxValue else configured.toInt + normalizeMaxLineLength(conf.get(History.EVENT_LOG_MAX_LINE_LENGTH)) + } + + private def normalizeMaxLineLength(configured: Long): Int = { + if (configured <= 0 || configured > MAX_LINE_LENGTH) MAX_LINE_LENGTH else configured.toInt } type ReplayEventsFilter = (String) => Boolean // utility filter that selects all event logs during replay val SELECT_ALL_FILTER: ReplayEventsFilter = { (eventString: String) => true } + + /** Accumulates a bounded prefix of one decoded line, allowing a possible trailing CR. */ + private[scheduler] class BoundedLineBuffer(maxLineLength: Int) { + private val buffer = new java.lang.StringBuilder() + private var byteLength = 0L + private val bufferLimit = maxLineLength.toLong + 1 + + def length: Int = buffer.length() + + def capacity: Int = buffer.capacity() + + def append(c: Char): Unit = { + if (byteLength <= bufferLimit) { + // The strict decoder emits valid surrogate pairs: count two bytes per half. + byteLength += (if (c < 0x80) 1 else if (c < 0x800 || Character.isSurrogate(c)) 2 else 3) + // Reserve one extra byte for a possible CR in the line ending. + if (byteLength <= bufferLimit) { + buffer.append(c) + } else { + // The line will be skipped; release the retained prefix while draining. + buffer.setLength(0) + buffer.trimToSize() + } + } + } + + def result(): Option[String] = { + val trailingCR = length > 0 && buffer.charAt(length - 1) == '\r' + val contentLength = byteLength - (if (trailingCR) 1 else 0) + if (contentLength > maxLineLength) { + None + } else if (trailingCR) { + Some(buffer.substring(0, length - 1)) + } else { + Some(buffer.toString) + } + } + } + } diff --git a/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala index f4652c23f63e2..f99a61eaa29f9 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala @@ -23,13 +23,17 @@ import java.util.concurrent.atomic.AtomicInteger import scala.collection.mutable.ArrayBuffer +import com.fasterxml.jackson.core.{JsonParseException, JsonProcessingException} +import com.fasterxml.jackson.databind.exc.MismatchedInputException import org.apache.hadoop.fs.Path +import org.apache.logging.log4j.Level import org.scalatest.BeforeAndAfter import org.apache.spark._ import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.deploy.history.EventLogFileReader import org.apache.spark.deploy.history.EventLogTestHelper._ +import org.apache.spark.internal.config.History import org.apache.spark.io.{CompressionCodec, LZ4CompressionCodec} import org.apache.spark.util.{JsonProtocol, JsonProtocolSuite, Utils} @@ -109,6 +113,221 @@ class ReplayListenerSuite extends SparkFunSuite with BeforeAndAfter with LocalSp assert(eventMonster.loggedEvents(1) === JsonProtocol.sparkEventToJsonString(applicationEnd)) } + test("SPARK-59804: Replay line limit uses UTF-8 bytes") { + val start = SparkListenerApplicationStart("x" * 6000, None, 125L, "user", None) + val json = JsonProtocol.sparkEventToJsonString(start) + assert(json.getBytes(StandardCharsets.UTF_8).length < 8 * 1024) + val listener = new EventBufferingListener + val conf = new SparkConf(false).set(History.EVENT_LOG_MAX_LINE_LENGTH.key, "8k") + val bus = new ReplayListenerBus(ReplayListenerBus.maxLineLength(conf)) + bus.addListener(listener) + val input = new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8)) + assert(bus.replay(input, "ascii")) + assert(listener.loggedEvents.toSeq == Seq(json)) + } + + test("SPARK-59804: Replay line limit handles raw UTF-8 boundaries and line endings") { + // scalastyle:off nonascii + val suffixes = Seq("x", "\u00e9", "\u4e2d", "\ud83d\ude00") + // scalastyle:on nonascii + for { + suffix <- suffixes + padding <- Seq(8, 8190, 8191, 8192) + ending <- Seq("\n", "\r\n", "") + delta <- -3 to 2 + } { + // Put the limit inside the final code point, and split UTF-8 sequences across read buffers. + val line = "x" * padding + suffix + val bytes = line.getBytes(StandardCharsets.UTF_8) + val limit = bytes.length + delta + val seen = new ArrayBuffer[String] + val bus = new ReplayListenerBus(limit) + val input = new ByteArrayInputStream((line + ending).getBytes(StandardCharsets.UTF_8)) + withClue(s"codePoint=${suffix.codePointAt(0)} padding=$padding " + + s"ending=${ending.map(_.toInt).mkString(",")} delta=$delta: ") { + // Observe raw lines before JSON parsing, without letting Jackson escape the test input. + assert(bus.replay(input, "utf8", eventsFilter = line => { + seen += line + false + })) + assert(seen.toSeq == (if (delta >= 0) Seq(line) else Seq.empty)) + } + } + } + + test("SPARK-59804: Replay line buffer stops retaining oversized content") { + for (limit <- Seq(0, 1, 8, 1024)) { + val buffer = new ReplayListenerBus.BoundedLineBuffer(limit) + for (_ <- 0 until 10000) { + buffer.append('x') + assert(buffer.length <= limit + 1) + } + assert(buffer.length == 0) + assert(buffer.capacity == 0) + assert(buffer.result().isEmpty) + } + } + + test("SPARK-59804: Replay retains a public no-argument constructor") { + val bus = classOf[ReplayListenerBus].getConstructor().newInstance() + assert(bus.effectiveMaxLineLength == ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH) + } + + test("SPARK-59804: Replay reports the physical location of skipped lines") { + val end = JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L)) + val input = new ByteArrayInputStream( + Seq(end, "x" * 2048, end, "x" * 2048).mkString("\n") + .getBytes(StandardCharsets.UTF_8)) + val appender = new LogAppender + appender.setThreshold(Level.DEBUG) + withLogAppender(appender, level = Some(Level.DEBUG)) { + assert(new ReplayListenerBus(1024).replay(input, "skipped-lines")) + } + val warnings = appender.loggingEvents.filter(_.getLevel == Level.WARN) + assert(warnings.size == 1) + assert(warnings.head.getMessage.getFormattedMessage.contains("first skipped line: 2")) + val debug = appender.loggingEvents.filter(_.getLevel == Level.DEBUG) + .map(_.getMessage.getFormattedMessage) + assert(debug.contains("Skipped event log line 2 in skipped-lines")) + assert(debug.contains("Skipped event log line 4 in skipped-lines")) + } + + test("SPARK-59804: Replay bounds malformed line diagnostics") { + val line = """{"Event":"SparkListenerJobStart","Job ID":1,"Stage IDs":[],""" + + """"Properties":{"bad":1},"padding":"""" + "x" * 10000 + "tail-marker\"}" + val appender = new LogAppender + withLogAppender(appender) { + assert(!new ReplayListenerBus().replay( + new ByteArrayInputStream(line.getBytes(StandardCharsets.UTF_8)), "semantic-error")) + } + val diagnostic = appender.loggingEvents.map(_.getMessage.getFormattedMessage) + .find(_.startsWith("Malformed line #1:")).get + assert(diagnostic.contains(s"line length: ${line.length} characters")) + assert(!diagnostic.contains("tail-marker")) + assert(diagnostic.length < 1200) + } + + test("SPARK-59804: Replay line limit defaults and maximum supported configuration") { + val conf = new SparkConf(false) + assert(conf.get(History.EVENT_LOG_MAX_LINE_LENGTH) == 256L * 1024 * 1024) + assert(ReplayListenerBus.maxLineLength(conf) == ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH) + for (value <- Seq("0", "-1", "512m", "536870913", "576m", "600m", "1g", + "2147483647", "3g")) { + conf.set(History.EVENT_LOG_MAX_LINE_LENGTH.key, value) + assert(ReplayListenerBus.maxLineLength(conf) == ReplayListenerBus.MAX_LINE_LENGTH) + } + } + + test("SPARK-59804: Replay constructor normalizes the effective line limit") { + val maximum = ReplayListenerBus.MAX_LINE_LENGTH + for (limit <- Seq(Int.MinValue, -1, 0, maximum, maximum + 1, Int.MaxValue)) { + val bus = new ReplayListenerBus(limit) + assert(bus.effectiveMaxLineLength == maximum) + val listener = new EventBufferingListener + bus.addListener(listener) + val json = JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L)) + assert(bus.replay(new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8)), + "normalized")) + assert(listener.loggedEvents.toSeq == Seq(json)) + } + for (limit <- Seq(1, 8192, ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH, maximum - 1)) { + assert(new ReplayListenerBus(limit).effectiveMaxLineLength == limit) + val conf = new SparkConf(false).set(History.EVENT_LOG_MAX_LINE_LENGTH, limit.toLong) + assert(ReplayListenerBus.maxLineLength(conf) == limit) + } + assert(new ReplayListenerBus().effectiveMaxLineLength == + ReplayListenerBus.DEFAULT_MAX_LINE_LENGTH) + } + + test("SPARK-59804: Replay preserves physical line numbers after skipping long lines") { + val end = JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L)) + val appender = new LogAppender + val bus = new ReplayListenerBus(1024) + val input = new ByteArrayInputStream( + (end + "\n" + "x" * 2048 + "\n" + "x" * 2048 + "\n{}\n") + .getBytes(StandardCharsets.UTF_8)) + withLogAppender(appender) { + assert(!bus.replay(input, "line-numbers")) + } + assert(appender.loggingEvents.exists(_.getMessage.getFormattedMessage + .contains("Malformed line #4: {}"))) + } + + test("SPARK-59804: Replay logs physical line numbers before rethrowing JSON errors") { + val end = JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L)) + val mappingError = + """{"Event":"org.apache.spark.util.TestListenerEvent","foo":"x","bar":[]}""" + for ((malformed, errorClass) <- Seq( + ("{bad", classOf[JsonParseException]), + (mappingError, classOf[MismatchedInputException]))) { + val appender = new LogAppender + val bus = new ReplayListenerBus(1024) + val input = new ByteArrayInputStream( + Seq(end, "x" * 2048, "x" * 2048, malformed, end).mkString("\n") + .getBytes(StandardCharsets.UTF_8)) + withLogAppender(appender) { + val error = intercept[JsonProcessingException] { + bus.replay(input, "json-errors", maybeTruncated = true) + } + assert(errorClass.isInstance(error)) + } + val diagnostics = appender.loggingEvents.filter(_.getMessage.getFormattedMessage + .contains("Exception parsing Spark event log: json-errors at line 4")) + assert(diagnostics.size == 1) + assert(diagnostics.head.getThrown == null) + } + } + + test("SPARK-59804: Replay propagates stream IO errors without reporting a stale parse line") { + val error = new IOException("read failure") + val end = JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L)) + val input = new ByteArrayInputStream((end + "\n").getBytes(StandardCharsets.UTF_8)) { + override def read(bytes: Array[Byte], offset: Int, length: Int): Int = { + if (available() == 0) throw error + super.read(bytes, offset, length) + } + } + val bus = new ReplayListenerBus() + val listener = new EventBufferingListener + bus.addListener(listener) + val appender = new LogAppender + withLogAppender(appender) { + assert(intercept[IOException] { + bus.replay(input, "read-error") + } eq error) + } + assert(listener.loggedEvents.toSeq == Seq(end)) + assert(!appender.loggingEvents.exists(_.getMessage.getFormattedMessage + .contains("Exception parsing Spark event log"))) + } + + test("SPARK-59804: Replay preserves line numbers for truncated logs and filtered iterators") { + val end = JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L)) + val truncated = "{\"Event\":" + val appender = new LogAppender + val bus = new ReplayListenerBus(1024) + val input = new ByteArrayInputStream( + ("x" * 2048 + "\n" + end + "\n" + truncated).getBytes(StandardCharsets.UTF_8)) + withLogAppender(appender) { + assert(bus.replay(input, "truncated", maybeTruncated = true, + eventsFilter = _ != end)) + assert(bus.replay(Iterator(end, end, truncated), "iterator", maybeTruncated = true, + eventsFilter = _ != end)) + } + val warnings = appender.loggingEvents.map(_.getMessage.getFormattedMessage) + .filter(_.contains("Got JsonParseException")) + assert(warnings.size == 2) + assert(warnings.forall(_.contains("at line 3,"))) + } + + test("SPARK-59804: Replay still rejects malformed UTF-8 in skipped lines") { + val bytes = Array.fill[Byte](20 * 1024)('x'.toByte) ++ Array(0xff.toByte, '\n'.toByte) + val bus = new ReplayListenerBus(1024) + intercept[java.nio.charset.MalformedInputException] { + bus.replay(new ByteArrayInputStream(bytes), "invalid-utf8") + } + } + /** * Test replaying compressed spark history file that internally throws an EOFException. To * avoid sensitivity to the compression specifics the test forces an EOFException to occur diff --git a/docs/monitoring.md b/docs/monitoring.md index 83fd85065f446..55cf5d7785093 100644 --- a/docs/monitoring.md +++ b/docs/monitoring.md @@ -440,12 +440,16 @@ Security options for the Spark History Server are covered more detail in the spark.history.fs.eventLog.maxLineLength - 512m + 256m - 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, which bounds 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.
+ Maximum UTF-8 byte length of a single event log line during replay, excluding the line + ending. Longer lines are skipped with a warning, which bounds 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. Buffer capacity grows + in steps: reducing 256m to 200m or 150m may not reduce the retained buffer; use 128m to + reach a smaller capacity. Values at or below + 0, or above 536870912, use the maximum supported limit of 536870912 bytes (512 MiB). + This cap avoids JVM array-size limits but does not guarantee sufficient heap space.
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.