[SPARK-11424] Guard against double-close() of RecordReaders

[JoshRosen]

TL;DR: We can rule out one rare but potential cause of input stream corruption via defensive programming.

Background

MAPREDUCE-5918 is a bug where an instance of a decompressor ends up getting placed into a pool multiple times. Since the pool is backed by a list instead of a set, this can lead to the same decompressor being used in different places at the same time, which is not safe because those decompressors will overwrite each other's buffers. Sometimes this buffer sharing will lead to exceptions but other times it will might silently result in invalid / garbled input.

That Hadoop bug is fixed in Hadoop 2.7 but is still present in many Hadoop versions that we wish to support. As a result, I think that we should try to work around this issue in Spark via defensive programming to prevent RecordReaders from being closed multiple times.

So far, I've had a hard time coming up with explanations of exactly how double-close()s occur in practice, but I do have a couple of explanations that work on paper.

For instance, it looks like #7424, added in 1.5, introduces at least one extremely~rare corner-case path where Spark could double-close() a LineRecordReader instance in a way that triggers the bug. Here are the steps involved in the bad execution that I brainstormed up:

• The task has finished reading input, so we call close().
• While handling the close call and trying to close the reader, reader.close() throws an exception
• We don't set reader = null after handling this exception, so the TaskCompletionListener also ends up calling NewHadoopRDD.close(), which, in turn, closes the record reader again.

In this hypothetical situation, LineRecordReader.close() could fail with an exception if its InputStream failed to close.
I googled for "Exception in RecordReader.close()" and it looks like it's possible for a closed Hadoop FileSystem to trigger an error there: SPARK-757, SPARK-2491

Looking at SPARK-3052, it seems like it's possible to get spurious exceptions there when there is an error reading from Hadoop. If the Hadoop FileSystem were to get into an error state right after reading the last record then it looks like we could hit the bug here in 1.5.

The fix

This patch guards against these issues by modifying HadoopRDD.close() and NewHadoopRDD.close() so that they set reader = null even if an exception occurs in the reader.close() call. In addition, I modified NextIterator. closeIfNeeded() to guard against double-close if the first close() call throws an exception.

I don't have an easy way to test this, since I haven't been able to reproduce the bug that prompted this patch, but these changes seem safe and seem to rule out the on-paper reproductions that I was able to brainstorm up.

[JoshRosen]

/cc @yhuai @marmbrus

[yhuai]

LGTM.

[harishreedharan]

+1.

[SparkQA]

Test build #44694 has finished for PR 9382 at commit 087aa63.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[JoshRosen]

Jenkins, retest this please.

[JoshRosen]

Flaky SQL test?

[SparkQA]

Test build #44703 has finished for PR 9382 at commit 087aa63.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[yhuai]

test this please

[SparkQA]

Test build #44716 has finished for PR 9382 at commit 087aa63.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[yhuai]

https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/44716/testReport/junit/org.apache.spark.sql/ColumnExpressionSuite/InputFileName/

Looks like a legitimate failure?

[JoshRosen]

@yhuai, here's the source code of the failing test:

  test("InputFileName") {
    withTempPath { dir =>
      val data = sparkContext.parallelize(0 to 10).toDF("id")
      data.write.parquet(dir.getCanonicalPath)
      val answer = sqlContext.read.parquet(dir.getCanonicalPath).select(inputFileName())
        .head.getString(0)
      assert(answer.contains(dir.getCanonicalPath))

      checkAnswer(data.select(inputFileName()).limit(1), Row(""))
    }
  }

Is the original assertion supposed to be checking that inputFileName() returns an empty string for data frames that were created by sc.parallelize()?

[JoshRosen]

Whoops, this is why the test was failing.

[JoshRosen]

Spotted the problem: I accidentally deleted a SqlNewHadoopRDD.unsetInputFileName() call.

[JoshRosen]

By the way, this use of nextIterator() explains why this bug was able to occur on 1.4.x: the first close() call, which throws an exception, comes from NextIterator, and the second comes from the task completion listener.

[JoshRosen]

Due to the NextIterator case in HadoopRDD that I pointed out up-thread, this bug potentially impacts all released versions of Spark. Therefore, I think we should fix this in 1.5.2, 1.6.0, 1.4.2, and 1.3.2.

[SparkQA]

Test build #44727 has finished for PR 9382 at commit 5ec97d7.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[JoshRosen]

Alright, fixed the tests and proofread this a second time, so I'm going to merge this to master and branch-1.5.