dongjoon-hyun commented on code in PR #59332:
URL: https://github.com/apache/spark/pull/59332#discussion_r4236859668
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -183,6 +183,86 @@ private[spark] object BasePythonRunner extends Logging {
ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
+ /** How often to log while a task completion listener waits for a pipelined
writer to exit. */
+ private val pipelinedWriterExitWarnIntervalMs = 10000L
+
+ /**
+ * Runs `writer` on [[pipelinedWriterThreadPool]] and registers a task
completion listener
+ * that interrupts it and blocks until it has exited.
+ *
+ * Listeners registered earlier run later (LIFO) and free memory backing the
input rows, e.g.
+ * the off-heap pages of a HybridRowQueue or of an off-heap vectorized
reader. If the writer
+ * were still serializing or buffering such a row at that point, it would be
a use-after-free
+ * that can crash the executor (SPARK-33277). `Future.get()` cannot be used
for this wait:
+ * once `cancel(true)` succeeds it throws `CancellationException`
immediately, without waiting
+ * for the interrupted runnable to return.
+ */
+ private[python] def startPipelinedWriter(writer: Runnable, context:
TaskContext): Unit = {
+ val writerTask = new PipelinedWriterTask(writer)
+ val writerFuture = pipelinedWriterThreadPool.submit(writerTask)
+ context.addTaskCompletionListener[Unit] { _ =>
+ writerTask.stopAndAwaitExit(writerFuture, taskIdentifier(context))
Review Comment:
Since the listener no longer looks at the `Future`, a fatal `Throwable`
escaping `PipelinedWriterRunnable.run` (`NonFatal` excludes
`VirtualMachineError`, `LinkageError`, etc.) is stored in the `FutureTask` and
silently dropped. In that case the writer also doesn't call `shutdownOutput()`
or `writer.setException(...)`, so the Python worker keeps waiting for input and
the reader waits for output with no root cause recorded. Since the listener now
knows the writer has exited, could we log the cause here (e.g. `future.get()`
after the latch, ignoring `CancellationException`)?
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -183,6 +183,86 @@ private[spark] object BasePythonRunner extends Logging {
ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
+ /** How often to log while a task completion listener waits for a pipelined
writer to exit. */
+ private val pipelinedWriterExitWarnIntervalMs = 10000L
+
+ /**
+ * Runs `writer` on [[pipelinedWriterThreadPool]] and registers a task
completion listener
+ * that interrupts it and blocks until it has exited.
+ *
+ * Listeners registered earlier run later (LIFO) and free memory backing the
input rows, e.g.
+ * the off-heap pages of a HybridRowQueue or of an off-heap vectorized
reader. If the writer
+ * were still serializing or buffering such a row at that point, it would be
a use-after-free
+ * that can crash the executor (SPARK-33277). `Future.get()` cannot be used
for this wait:
+ * once `cancel(true)` succeeds it throws `CancellationException`
immediately, without waiting
+ * for the interrupted runnable to return.
+ */
+ private[python] def startPipelinedWriter(writer: Runnable, context:
TaskContext): Unit = {
+ val writerTask = new PipelinedWriterTask(writer)
+ val writerFuture = pipelinedWriterThreadPool.submit(writerTask)
+ context.addTaskCompletionListener[Unit] { _ =>
Review Comment:
This orders the writer listener only against the listeners registered
*before* this line. In pipelined mode, the upstream iterator is first pulled on
the writer pool thread, so any completion listener that upstream operators
register lazily is pushed *after* this one and runs *before* it under LIFO,
while the writer is still running and not even interrupted yet. For example:
- DSv2 Parquet: `ParquetReaderCallback.initIfNotAlready` registers
`closeCurrent()` when the first file is opened
(`ParquetPartitionReaderFactory.scala` L391), which closes the vectorized
reader (off-heap column vectors when
`spark.sql.columnVector.offheap.enabled=true`).
- `ExternalAppendOnlyUnsafeRowArray` (SMJ / Window upstream) creates an
`UnsafeExternalSorter` lazily once the threshold is exceeded, and its
constructor registers `cleanupResources()`.
In both cases, a `LIMIT` that completes the task frees the memory under the
running writer, which is the same use-after-free that this PR is trying to
prevent. The new tests register the cleanup listener on the task thread before
the writer starts, so they don't cover this. Could we stop and wait for the
writer before *any* completion listener runs (e.g. a pre-completion hook in
`TaskContextImpl`), instead of relying on the position in the LIFO stack?
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -183,6 +183,86 @@ private[spark] object BasePythonRunner extends Logging {
ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
+ /** How often to log while a task completion listener waits for a pipelined
writer to exit. */
+ private val pipelinedWriterExitWarnIntervalMs = 10000L
+
+ /**
+ * Runs `writer` on [[pipelinedWriterThreadPool]] and registers a task
completion listener
+ * that interrupts it and blocks until it has exited.
+ *
+ * Listeners registered earlier run later (LIFO) and free memory backing the
input rows, e.g.
+ * the off-heap pages of a HybridRowQueue or of an off-heap vectorized
reader. If the writer
+ * were still serializing or buffering such a row at that point, it would be
a use-after-free
+ * that can crash the executor (SPARK-33277). `Future.get()` cannot be used
for this wait:
+ * once `cancel(true)` succeeds it throws `CancellationException`
immediately, without waiting
+ * for the interrupted runnable to return.
+ */
+ private[python] def startPipelinedWriter(writer: Runnable, context:
TaskContext): Unit = {
+ val writerTask = new PipelinedWriterTask(writer)
+ val writerFuture = pipelinedWriterThreadPool.submit(writerTask)
+ context.addTaskCompletionListener[Unit] { _ =>
+ writerTask.stopAndAwaitExit(writerFuture, taskIdentifier(context))
+ }
+ }
+
+ /**
+ * Wraps a pipelined writer so that a task completion listener can wait
until the writer has
+ * either exited or is guaranteed never to run.
+ */
+ private[python] class PipelinedWriterTask(writer: Runnable) extends Runnable
{
+ // Set by whichever of run() and stopAndAwaitExit() gets here first. If
run() wins, the
+ // writer runs and stopAndAwaitExit() waits for it; otherwise run() skips
the writer.
+ private val claimed = new AtomicBoolean(false)
+ private val exited = new CountDownLatch(1)
+
+ override def run(): Unit = {
+ if (claimed.compareAndSet(false, true)) {
+ try {
+ writer.run()
+ } finally {
+ exited.countDown()
+ }
+ }
+ }
+
+ def stopAndAwaitExit(future: Future[_], taskName: String): Unit = {
+ // Interrupts the writer thread if the writer is running. This unblocks
channel.write
+ // (the JDK closes the channel and throws ClosedByInterruptException),
and the writer
+ // loop checks the interrupt flag between rows, so the wait below is
normally short.
+ future.cancel(true)
+ if (claimed.compareAndSet(false, true)) {
+ // The writer never started and now never will.
+ return
+ }
+ // Wait without a time bound: giving up would let the listeners that run
next free memory
+ // the writer may still be reading, and failing this listener would not
stop them either.
+ // A killed task that stays stuck here is handled by the task reaper, if
enabled.
+ val startNs = System.nanoTime()
Review Comment:
nit. The `interrupted` flag, the outer `try`/`finally`, and the inner `catch
InterruptedException` re-implement Guava's
`Uninterruptibles.awaitUninterruptibly(CountDownLatch, long, TimeUnit)`, which
also restores the caller's interrupt status before returning. `core` already
uses `Uninterruptibles` (e.g. `AsyncEventQueue.scala`). This could be
simplified to something like:
```scala
while (!Uninterruptibles.awaitUninterruptibly(
exited, pipelinedWriterExitWarnIntervalMs, TimeUnit.MILLISECONDS)) {
logWarning(...)
}
```
##########
core/src/test/scala/org/apache/spark/api/python/BasePythonRunnerSuite.scala:
##########
@@ -50,6 +50,105 @@ class BasePythonRunnerSuite extends SparkFunSuite {
}
}
+ /**
+ * A writer that, once started, keeps running until `release` is counted
down, ignoring
+ * interrupts the way a writer in the middle of copying a row into a
HybridRowQueue page does.
+ */
+ private class UninterruptibleWriter extends Runnable {
+ val started = new CountDownLatch(1)
+ val release = new CountDownLatch(1)
+ @volatile var exited = false
+
+ override def run(): Unit = {
+ started.countDown()
+ try {
+ var done = false
+ while (!done) {
+ try {
+ done = release.await(1, TimeUnit.MINUTES)
+ } catch {
+ case _: InterruptedException => // keep going, like a row copy in
progress
+ }
+ }
+ } finally {
+ exited = true
+ }
+ }
+ }
+
+ test("SPARK-60113: task completion waits for the pipelined writer to exit") {
+ val context = TaskContext.empty()
+ val writer = new UninterruptibleWriter
+ // Registered before the writer, so it runs after the writer's listener
(LIFO), like the
+ // listener that closes the HybridRowQueue and frees its pages.
+ @volatile var exitedBeforeCleanup: Option[Boolean] = None
+ context.addTaskCompletionListener[Unit] { _ => exitedBeforeCleanup =
Some(writer.exited) }
+ BasePythonRunner.startPipelinedWriter(writer, context)
+ assert(writer.started.await(1, TimeUnit.MINUTES))
+
+ val completion = new Thread(() => context.markTaskCompleted(None))
+ completion.setDaemon(true)
+ completion.start()
+ try {
+ // Task completion must stay blocked while the writer is still running.
+ completion.join(500)
Review Comment:
`completion.isAlive` and `exitedBeforeCleanup.isEmpty` are also true when
the completion thread has not reached the writer listener yet. On a slow CI
machine where the thread isn't scheduled within 500 ms, this test would pass
even with the old `cancel(true)` + `get()` listener. How about registering a
listener *after* `startPipelinedWriter` (which runs first under LIFO) that
counts down a latch, and waiting on it before checking `isAlive`? The same
applies to `completion.join(200)` + `interrupt()` at L121, which doesn't verify
that the interrupt arrives during the wait.
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -183,6 +183,86 @@ private[spark] object BasePythonRunner extends Logging {
ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
+ /** How often to log while a task completion listener waits for a pipelined
writer to exit. */
+ private val pipelinedWriterExitWarnIntervalMs = 10000L
+
+ /**
+ * Runs `writer` on [[pipelinedWriterThreadPool]] and registers a task
completion listener
+ * that interrupts it and blocks until it has exited.
+ *
+ * Listeners registered earlier run later (LIFO) and free memory backing the
input rows, e.g.
+ * the off-heap pages of a HybridRowQueue or of an off-heap vectorized
reader. If the writer
+ * were still serializing or buffering such a row at that point, it would be
a use-after-free
+ * that can crash the executor (SPARK-33277). `Future.get()` cannot be used
for this wait:
+ * once `cancel(true)` succeeds it throws `CancellationException`
immediately, without waiting
+ * for the interrupted runnable to return.
+ */
+ private[python] def startPipelinedWriter(writer: Runnable, context:
TaskContext): Unit = {
+ val writerTask = new PipelinedWriterTask(writer)
+ val writerFuture = pipelinedWriterThreadPool.submit(writerTask)
+ context.addTaskCompletionListener[Unit] { _ =>
+ writerTask.stopAndAwaitExit(writerFuture, taskIdentifier(context))
+ }
+ }
+
+ /**
+ * Wraps a pipelined writer so that a task completion listener can wait
until the writer has
+ * either exited or is guaranteed never to run.
+ */
+ private[python] class PipelinedWriterTask(writer: Runnable) extends Runnable
{
+ // Set by whichever of run() and stopAndAwaitExit() gets here first. If
run() wins, the
+ // writer runs and stopAndAwaitExit() waits for it; otherwise run() skips
the writer.
+ private val claimed = new AtomicBoolean(false)
+ private val exited = new CountDownLatch(1)
+
+ override def run(): Unit = {
+ if (claimed.compareAndSet(false, true)) {
+ try {
+ writer.run()
+ } finally {
+ exited.countDown()
+ }
+ }
+ }
+
+ def stopAndAwaitExit(future: Future[_], taskName: String): Unit = {
+ // Interrupts the writer thread if the writer is running. This unblocks
channel.write
+ // (the JDK closes the channel and throws ClosedByInterruptException),
and the writer
+ // loop checks the interrupt flag between rows, so the wait below is
normally short.
+ future.cancel(true)
+ if (claimed.compareAndSet(false, true)) {
+ // The writer never started and now never will.
+ return
+ }
+ // Wait without a time bound: giving up would let the listeners that run
next free memory
+ // the writer may still be reading, and failing this listener would not
stop them either.
+ // A killed task that stays stuck here is handled by the task reaper, if
enabled.
Review Comment:
This unbounded wait is a behavior change for writers blocked in upstream I/O
that does not respond to interrupts, e.g. JDBC `ResultSet.next()` on a slow
fetch, or an HTTP/S3 stream read with a long (or no) socket timeout. When a
`LIMIT` is satisfied, the task completes normally, `cancel(true)` cannot
unblock the read, and `markTaskCompleted` now waits until that read returns,
holding the task slot and only logging a warning every 10 seconds. Previously
the listener returned immediately.
Also, the task reaper only applies to killed tasks and
`spark.task.reaper.enabled` is `false` by default, so nothing recovers the
normal completion case. In these examples the rows are on-heap, so the wait
doesn't buy any safety. Could we mention this trade-off explicitly, or consider
limiting the wait to the cases where the input memory can actually be freed
under the writer?
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -183,6 +183,86 @@ private[spark] object BasePythonRunner extends Logging {
ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
+ /** How often to log while a task completion listener waits for a pipelined
writer to exit. */
+ private val pipelinedWriterExitWarnIntervalMs = 10000L
+
+ /**
+ * Runs `writer` on [[pipelinedWriterThreadPool]] and registers a task
completion listener
+ * that interrupts it and blocks until it has exited.
+ *
+ * Listeners registered earlier run later (LIFO) and free memory backing the
input rows, e.g.
+ * the off-heap pages of a HybridRowQueue or of an off-heap vectorized
reader. If the writer
+ * were still serializing or buffering such a row at that point, it would be
a use-after-free
+ * that can crash the executor (SPARK-33277). `Future.get()` cannot be used
for this wait:
+ * once `cancel(true)` succeeds it throws `CancellationException`
immediately, without waiting
+ * for the interrupted runnable to return.
+ */
+ private[python] def startPipelinedWriter(writer: Runnable, context:
TaskContext): Unit = {
+ val writerTask = new PipelinedWriterTask(writer)
+ val writerFuture = pipelinedWriterThreadPool.submit(writerTask)
+ context.addTaskCompletionListener[Unit] { _ =>
+ writerTask.stopAndAwaitExit(writerFuture, taskIdentifier(context))
+ }
+ }
+
+ /**
+ * Wraps a pipelined writer so that a task completion listener can wait
until the writer has
+ * either exited or is guaranteed never to run.
+ */
+ private[python] class PipelinedWriterTask(writer: Runnable) extends Runnable
{
+ // Set by whichever of run() and stopAndAwaitExit() gets here first. If
run() wins, the
+ // writer runs and stopAndAwaitExit() waits for it; otherwise run() skips
the writer.
+ private val claimed = new AtomicBoolean(false)
+ private val exited = new CountDownLatch(1)
+
+ override def run(): Unit = {
+ if (claimed.compareAndSet(false, true)) {
+ try {
+ writer.run()
+ } finally {
+ exited.countDown()
+ }
+ }
+ }
+
+ def stopAndAwaitExit(future: Future[_], taskName: String): Unit = {
+ // Interrupts the writer thread if the writer is running. This unblocks
channel.write
+ // (the JDK closes the channel and throws ClosedByInterruptException),
and the writer
+ // loop checks the interrupt flag between rows, so the wait below is
normally short.
Review Comment:
nit. `checks the interrupt flag between rows` is not accurate. The loop
checks it only between `writeNextInputToStream` calls. For Arrow runners, one
call serializes a whole batch (up to `arrowMaxRecordsPerBatch` rows) and pulls
from upstream, and `ColumnarArrowEvalPythonEvaluatorFactory` Path 2 copies all
rows of an input `ColumnarBatch` into the `HybridRowQueue` in one step without
checking the interrupt. The same wording is in the PR description. Since the
wait below is unbounded, it would be good to describe it precisely.
##########
core/src/test/scala/org/apache/spark/api/python/BasePythonRunnerSuite.scala:
##########
@@ -50,6 +50,105 @@ class BasePythonRunnerSuite extends SparkFunSuite {
}
}
+ /**
+ * A writer that, once started, keeps running until `release` is counted
down, ignoring
+ * interrupts the way a writer in the middle of copying a row into a
HybridRowQueue page does.
+ */
+ private class UninterruptibleWriter extends Runnable {
+ val started = new CountDownLatch(1)
+ val release = new CountDownLatch(1)
+ @volatile var exited = false
+
+ override def run(): Unit = {
+ started.countDown()
+ try {
+ var done = false
+ while (!done) {
+ try {
+ done = release.await(1, TimeUnit.MINUTES)
+ } catch {
+ case _: InterruptedException => // keep going, like a row copy in
progress
+ }
+ }
+ } finally {
+ exited = true
+ }
+ }
+ }
+
+ test("SPARK-60113: task completion waits for the pipelined writer to exit") {
+ val context = TaskContext.empty()
+ val writer = new UninterruptibleWriter
+ // Registered before the writer, so it runs after the writer's listener
(LIFO), like the
+ // listener that closes the HybridRowQueue and frees its pages.
+ @volatile var exitedBeforeCleanup: Option[Boolean] = None
+ context.addTaskCompletionListener[Unit] { _ => exitedBeforeCleanup =
Some(writer.exited) }
+ BasePythonRunner.startPipelinedWriter(writer, context)
+ assert(writer.started.await(1, TimeUnit.MINUTES))
+
+ val completion = new Thread(() => context.markTaskCompleted(None))
+ completion.setDaemon(true)
+ completion.start()
+ try {
+ // Task completion must stay blocked while the writer is still running.
+ completion.join(500)
+ assert(completion.isAlive)
+ assert(exitedBeforeCleanup.isEmpty)
+ } finally {
+ writer.release.countDown()
+ }
+ completion.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!completion.isAlive)
+ assert(exitedBeforeCleanup === Some(true))
+ }
+
+ test("SPARK-60113: task completion waits for the pipelined writer even if
interrupted") {
+ val context = TaskContext.empty()
+ val writer = new UninterruptibleWriter
+ @volatile var exitedBeforeCleanup: Option[Boolean] = None
+ context.addTaskCompletionListener[Unit] { _ => exitedBeforeCleanup =
Some(writer.exited) }
+ BasePythonRunner.startPipelinedWriter(writer, context)
+ assert(writer.started.await(1, TimeUnit.MINUTES))
+
+ @volatile var interruptedAfterCompletion = false
+ val completion = new Thread(() => {
+ context.markTaskCompleted(None)
+ interruptedAfterCompletion = Thread.currentThread().isInterrupted
+ })
+ completion.setDaemon(true)
+ completion.start()
+ try {
+ completion.join(200)
+ // E.g. a task kill interrupting the task thread while it runs the
listeners.
+ completion.interrupt()
+ completion.join(500)
+ assert(completion.isAlive)
+ assert(exitedBeforeCleanup.isEmpty)
+ } finally {
+ writer.release.countDown()
+ }
+ completion.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!completion.isAlive)
+ assert(exitedBeforeCleanup === Some(true))
+ // The interrupt is preserved for the code that runs after the listeners.
+ assert(interruptedAfterCompletion)
+ }
+
+ test("SPARK-60113: a pipelined writer stopped before it starts never runs") {
+ val writer = new UninterruptibleWriter
+ val writerTask = new BasePythonRunner.PipelinedWriterTask(writer)
+ // A future the pool has not started yet: stopping must not wait for it.
+ val future = new FutureTask[Unit](writerTask, ())
Review Comment:
This `FutureTask` is never run by a pool thread, so the test covers only the
case where `FutureTask.run` sees the cancelled state and never calls
`PipelinedWriterTask.run`. The real race isn't covered: the pool thread has
already entered `FutureTask.run`, `cancel(true)` interrupts it, and `run()`
still wins the claim and starts the writer with the interrupt flag already set.
In that case the listener must wait for the latch, so a test for it would
protect against a future change that breaks the countdown on that path.
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -183,6 +183,86 @@ private[spark] object BasePythonRunner extends Logging {
ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
+ /** How often to log while a task completion listener waits for a pipelined
writer to exit. */
+ private val pipelinedWriterExitWarnIntervalMs = 10000L
+
+ /**
+ * Runs `writer` on [[pipelinedWriterThreadPool]] and registers a task
completion listener
+ * that interrupts it and blocks until it has exited.
+ *
+ * Listeners registered earlier run later (LIFO) and free memory backing the
input rows, e.g.
+ * the off-heap pages of a HybridRowQueue or of an off-heap vectorized
reader. If the writer
+ * were still serializing or buffering such a row at that point, it would be
a use-after-free
+ * that can crash the executor (SPARK-33277). `Future.get()` cannot be used
for this wait:
+ * once `cancel(true)` succeeds it throws `CancellationException`
immediately, without waiting
+ * for the interrupted runnable to return.
+ */
+ private[python] def startPipelinedWriter(writer: Runnable, context:
TaskContext): Unit = {
+ val writerTask = new PipelinedWriterTask(writer)
+ val writerFuture = pipelinedWriterThreadPool.submit(writerTask)
+ context.addTaskCompletionListener[Unit] { _ =>
+ writerTask.stopAndAwaitExit(writerFuture, taskIdentifier(context))
+ }
+ }
+
+ /**
+ * Wraps a pipelined writer so that a task completion listener can wait
until the writer has
+ * either exited or is guaranteed never to run.
+ */
+ private[python] class PipelinedWriterTask(writer: Runnable) extends Runnable
{
+ // Set by whichever of run() and stopAndAwaitExit() gets here first. If
run() wins, the
+ // writer runs and stopAndAwaitExit() waits for it; otherwise run() skips
the writer.
+ private val claimed = new AtomicBoolean(false)
+ private val exited = new CountDownLatch(1)
+
+ override def run(): Unit = {
+ if (claimed.compareAndSet(false, true)) {
+ try {
+ writer.run()
+ } finally {
+ exited.countDown()
+ }
+ }
+ }
+
+ def stopAndAwaitExit(future: Future[_], taskName: String): Unit = {
+ // Interrupts the writer thread if the writer is running. This unblocks
channel.write
+ // (the JDK closes the channel and throws ClosedByInterruptException),
and the writer
+ // loop checks the interrupt flag between rows, so the wait below is
normally short.
+ future.cancel(true)
+ if (claimed.compareAndSet(false, true)) {
+ // The writer never started and now never will.
+ return
+ }
+ // Wait without a time bound: giving up would let the listeners that run
next free memory
+ // the writer may still be reading, and failing this listener would not
stop them either.
+ // A killed task that stays stuck here is handled by the task reaper, if
enabled.
+ val startNs = System.nanoTime()
+ var interrupted = false
+ try {
+ var done = false
+ while (!done) {
+ try {
+ done = exited.await(pipelinedWriterExitWarnIntervalMs,
TimeUnit.MILLISECONDS)
+ if (!done) {
+ val waitedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNs)
+ logWarning(log"Still waiting for the pipelined Python writer of
" +
+ log"${MDC(TASK_NAME, taskName)} to exit after " +
+ log"${MDC(TOTAL_TIME, waitedMs)} ms; task completion cleanup
is blocked " +
+ log"until it does.")
+ }
+ } catch {
+ case _: InterruptedException => interrupted = true
+ }
+ }
+ } finally {
+ if (interrupted) {
Review Comment:
Restoring the interrupt flag here means that every listener registered
earlier (which runs after this one) runs with the flag set, e.g.
`worker.stop()` (L560), `queue.close()` of the `HybridRowQueue` (which deletes
spill files), and the reader/allocator closes. Since this listener can now take
much longer, a task-kill interrupt is more likely to land during this wait and
be carried over. Any blocking NIO channel operation or lock wait in those
listeners would then fail with `ClosedByInterruptException` /
`InterruptedException`, and the cleanup would be skipped or logged as a
listener failure. Previously, the old listener didn't wait at all. Is this
intended? If so, could we add a comment about the impact on the following
listeners? The new test asserts the propagation, so it would lock in this
behavior.
--
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]
