viirya commented on code in PR #59074:
URL: https://github.com/apache/spark/pull/59074#discussion_r4117452237
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +109,85 @@ class ReplayListenerSuite extends SparkFunSuite with
BeforeAndAfter with LocalSp
assert(eventMonster.loggedEvents(1) ===
JsonProtocol.sparkEventToJsonString(applicationEnd))
}
+ test("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 bus = new ReplayListenerBus(8 * 1024)
+ bus.addListener(listener)
+ val input = new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8))
+ assert(bus.replay(input, "ascii"))
+ assert(listener.loggedEvents.toSeq == Seq(json))
+ }
+
+ test("Replay line limit handles UTF-8 boundaries and line endings") {
+ val end =
JsonProtocol.sparkEventToJsonString(SparkListenerApplicationEnd(1000L))
+ // scalastyle:off nonascii
+ val names = Seq("x", "\u00e9", "\u4e2d", "\ud83d\ude00")
+ // scalastyle:on nonascii
+ for {
+ name <- names
+ repetitions <- Seq(100, 4097)
+ ending <- Seq("\n", "\r\n", "")
+ excess <- Seq(0, 1)
+ } {
+ val start = SparkListenerApplicationStart(name * repetitions, None,
125L, "user", None)
+ val json = JsonProtocol.sparkEventToJsonString(start)
+ val limit = json.getBytes(StandardCharsets.UTF_8).length - excess
+ val listener = new EventBufferingListener
+ val bus = new ReplayListenerBus(limit)
+ bus.addListener(listener)
+ // Put a short event first, including when the final line has no
terminator.
+ val input = new ByteArrayInputStream((end + "\n" + json + ending)
+ .getBytes(StandardCharsets.UTF_8))
+ assert(bus.replay(input, "utf8"))
+ val expected = if (excess == 0) Seq(end, json) else Seq(end)
+ assert(listener.loggedEvents.toSeq == expected, s"name=$name
ending=$ending excess=$excess")
Review Comment:
Updated in ab7c85da8f6. Both replay success and observed-line assertions are
wrapped in a clue containing the code point, padding, numeric line-ending bytes
and limit delta. The old repetitions grid is replaced by deliberate boundary
offsets.
I retained the short ASCII application-event test because it now also
exercises the `8k` configuration and successful JSON event delivery. The raw
UTF-8 grid deliberately bypasses JSON parsing so serialization cannot hide the
byte sequences being tested.
##########
core/src/test/scala/org/apache/spark/scheduler/ReplayListenerSuite.scala:
##########
@@ -109,6 +109,85 @@ 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:
Confirmed by rerunning all five tests from the initial PR revision against
the original implementation: four fail, and the malformed-UTF-8 test passes.
The earlier count of three came from a run before the truncated/filtered test
was added. I removed the stale count from the PR description.
The revised suite has 15 passing tests, and both production and test
scalastyle checks pass.
--
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]
