Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11628#discussion_r56627433
--- Diff: core/src/main/scala/org/apache/spark/rdd/PipedRDD.scala ---
@@ -118,63 +120,94 @@ private[spark] class PipedRDD[T: ClassTag](
val proc = pb.start()
val env = SparkEnv.get
+ val childThreadException = new AtomicReference[Throwable](null)
// Start a thread to print the process's stderr to ours
- new Thread("stderr reader for " + command) {
- override def run() {
- for (line <- Source.fromInputStream(proc.getErrorStream).getLines)
{
- // scalastyle:off println
- System.err.println(line)
- // scalastyle:on println
+ new Thread(s"stderr reader for $command") {
+ override def run(): Unit = {
+ val err = proc.getErrorStream
+ try {
+ for (line <- Source.fromInputStream(err).getLines) {
+ // scalastyle:off println
+ System.err.println(line)
+ // scalastyle:on println
+ }
+ } catch {
+ case t: Throwable => childThreadException.set(t)
+ } finally {
+ err.close()
}
}
}.start()
// Start a thread to feed the process input from our parent's iterator
- new Thread("stdin writer for " + command) {
- override def run() {
+ new Thread(s"stdin writer for $command") {
+ override def run(): Unit = {
TaskContext.setTaskContext(context)
val out = new PrintWriter(proc.getOutputStream)
-
- // scalastyle:off println
- // input the pipe context firstly
- if (printPipeContext != null) {
- printPipeContext(out.println(_))
- }
- for (elem <- firstParent[T].iterator(split, context)) {
- if (printRDDElement != null) {
- printRDDElement(elem, out.println(_))
- } else {
- out.println(elem)
+ try {
+ // scalastyle:off println
+ // input the pipe context firstly
+ if (printPipeContext != null) {
+ printPipeContext(out.println)
+ }
+ for (elem <- firstParent[T].iterator(split, context)) {
+ if (printRDDElement != null) {
+ printRDDElement(elem, out.println)
+ } else {
+ out.println(elem)
+ }
}
+ // scalastyle:on println
+ } catch {
+ case t: Throwable => childThreadException.set(t)
+ } finally {
+ out.close()
--- End diff --
Hm, really? why would closing the stream _necessarily_ cause it to exit
with status 0?
It's important to try to close in the exception case; in the normal case,
this closes when the stream has no more to return. I agree that's not
necessarily the moment the process is completed (I think?) but closing the
stream does not kill the process.
You're referring to `Thread.destroy` here and below, but we are calling
`Process.destroy`. I don't think we can destroy the process when _one_ stream
is done since the other might not be. I suppose I can see a more complex
mechanism to wait until both finish and then close both, but I am still not
sure that's actually a problem?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
