dongjoon-hyun commented on code in PR #59568:
URL: https://github.com/apache/spark/pull/59568#discussion_r4307965026
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/streaming/PythonForeachWriterSuite.scala:
##########
@@ -66,6 +68,66 @@ class PythonForeachWriterSuite extends SparkFunSuite with
Eventually with Mockit
b.assertOutput(1 to 2000)
}
+ test("SPARK-60310: UnsafeRowBuffer: close does not free a row that the
reading thread uses") {
+ val buffer = createBuffer(4 << 10)
+ val intProj = UnsafeProjection.create(Array[DataType](IntegerType))
+ buffer.add(intProj(new GenericInternalRow(Array[Any](1))))
+ val rowRead = new CountDownLatch(1)
+ val bufferClosed = new CountDownLatch(1)
+ @volatile var value = 0
+ // Like the WriterThread converting a row while the task completion
listener closes the
+ // buffer, e.g. after the task was killed while waiting for the
WriterThread.
+ val reader = new Thread() {
+ override def run(): Unit = {
+ val row = buffer.iterator.next()
+ rowRead.countDown()
+ bufferClosed.await()
+ value = row.getInt(0)
+ }
+ }
+ reader.start()
+ try {
+ assert(rowRead.await(60, TimeUnit.SECONDS))
+ buffer.close()
+ } finally {
+ bufferClosed.countDown()
+ reader.join()
+ }
+ // A row in a freed page reads 0x5a5a5a5a, the pattern of
spark.memory.debugFill.
Review Comment:
Thank you. I updated the comment.
--
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]
