[SPARK-23399][SQL] Register a task completion listener first for OrcColumnarBatchReader #20590

dongjoon-hyun · 2018-02-12T21:55:28Z

What changes were proposed in this pull request?

This PR aims to resolve an open file leakage issue reported at SPARK-23390 by moving the listener registration position. Currently, the sequence is like the following.

Create batchReader
batchReader.initialize opens a ORC file.
batchReader.initBatch may take a long time to alloc memory in some environment and cause errors.
Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => iter.close()))

This PR moves 4 before 2 and 3. To sum up, the new sequence is 1 -> 4 -> 2 -> 3.

How was this patch tested?

Manual. The following test case makes OOM intentionally to cause leaked filesystem connection in the current code base. With this patch, leakage doesn't occurs.

  // This should be tested manually because it raises OOM intentionally
  // in order to cause `Leaked filesystem connection`.
  test("SPARK-23399 Register a task completion listener first for OrcColumnarBatchReader") {
    withSQLConf(SQLConf.ORC_VECTORIZED_READER_BATCH_SIZE.key -> s"${Int.MaxValue}") {
      withTempDir { dir =>
        val basePath = dir.getCanonicalPath
        Seq(0).toDF("a").write.format("orc").save(new Path(basePath, "first").toString)
        Seq(1).toDF("a").write.format("orc").save(new Path(basePath, "second").toString)
        val df = spark.read.orc(
          new Path(basePath, "first").toString,
          new Path(basePath, "second").toString)
        val e = intercept[SparkException] {
          df.collect()
        }
        assert(e.getCause.isInstanceOf[OutOfMemoryError])
      }
    }
  }

…lumnarBatchReader

dongjoon-hyun · 2018-02-12T21:57:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala

+          val iter = new RecordReaderIterator(batchReader)
+          Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => iter.close()))
+
          batchReader.initialize(fileSplit, taskAttemptContext)


According to the reported case, the ORC file is opened here.
But, it seems that the task is killed, TaskKilled (Stage cancelled), during initBatch before registering its listener . For a case throwing Exception at initBatch, this PR prevents the open file leakage.

@cloud-fan and @gatorsmile . Could you take a look this?
For ORC library, it looks okay when we call close correctly.

@dongjoon-hyun Thanks for this fix! My question is how do we know if close is not called before and is called now? Have you verified it?

Because I tried to verify it manually in local, seems close is called before this change. Maybe I miss something or this is environment depending.

SparkQA · 2018-02-13T00:07:32Z

Test build #87349 has finished for PR 20590 at commit 198f186.

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

kiszk · 2018-02-13T01:14:59Z

Umm...

kiszk · 2018-02-13T01:54:11Z

@dongjoon-hyun One question. Do you know which test case in FileBasedDataSourceSuite may cause this failure? Now, we are seeing a stack trace at afterEach.

cloud-fan · 2018-02-13T02:50:13Z

looks reasonable.

batchReader.initBatch throw FileNotException, and we enter afterEach, detect the file stream leak and fail.

cloud-fan · 2018-02-13T02:51:40Z

I know it's hard to add a test, we need a malformed ORC file to make the reader fail midway. @dongjoon-hyun do you think it's possible to generate such a ORC file?

dongjoon-hyun · 2018-02-13T06:32:52Z

Thank you for review, @viirya , @kiszk , @cloud-fan .
Yep. I'm still trying to reproduce it by a test case. I'll inform you later.

dongjoon-hyun · 2018-02-13T09:28:23Z

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala

+        assert(e.getCause.isInstanceOf[OutOfMemoryError])
+      }
+    }
+  }


Hi, All.
The above test case generates the same leakage reported in JIRA.
And, this PR fixes that. Please try this test case in IntelliJ with the master branch.

cloud-fan · 2018-02-13T10:56:20Z

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcSourceSuite.scala

  }
+
+  // This should be tested manually because it raises OOM intentionally
+  // in order to cause `Leaked filesystem connection`. The test suite dies, too.


ah, nice trick to fail the reader midway!

But it's a little weird to have it as a unit test, shall we just put it in the PR description and say it's manually tested? This test needs to be run manually anyway...

SparkQA · 2018-02-13T11:37:51Z

Test build #87384 has finished for PR 20590 at commit 3b8cb0a.

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

SparkQA · 2018-02-13T12:33:59Z

Test build #87385 has finished for PR 20590 at commit d4cc32e.

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

kiszk · 2018-02-13T12:51:20Z

retest this please

kiszk · 2018-02-13T15:43:05Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala

          val batchReader = new OrcColumnarBatchReader(
            enableOffHeapColumnVector && taskContext.isDefined, copyToSpark, capacity)
+          val iter = new RecordReaderIterator(batchReader)
+          Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => iter.close()))


Could you please add comment why we put this registration here with SPARK-23399. Since we would forget this investigation in the future :), this comment will help us and will remind to run the test case manually.

SparkQA · 2018-02-13T15:52:52Z

Test build #87393 has finished for PR 20590 at commit d4cc32e.

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

SparkQA · 2018-02-13T18:38:24Z

Test build #87397 has finished for PR 20590 at commit 5a9fa0b.

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

SparkQA · 2018-02-13T19:13:08Z

Test build #87402 has finished for PR 20590 at commit 18c2485.

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

dongjoon-hyun · 2018-02-13T20:37:54Z

Retest this please.

SparkQA · 2018-02-14T00:02:37Z

Test build #87425 has finished for PR 20590 at commit 18c2485.

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

dongjoon-hyun · 2018-02-14T00:07:53Z

Hi, @cloud-fan and @gatorsmile .
It's ready for review again. I think this definitely will reduce the flakiness for that suite.
But, I'm still looking at some other places, too.

cloud-fan · 2018-02-14T02:57:07Z

thanks, merging to master/2.3!

…olumnarBatchReader This PR aims to resolve an open file leakage issue reported at [SPARK-23390](https://issues.apache.org/jira/browse/SPARK-23390) by moving the listener registration position. Currently, the sequence is like the following. 1. Create `batchReader` 2. `batchReader.initialize` opens a ORC file. 3. `batchReader.initBatch` may take a long time to alloc memory in some environment and cause errors. 4. `Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => iter.close()))` This PR moves 4 before 2 and 3. To sum up, the new sequence is 1 -> 4 -> 2 -> 3. Manual. The following test case makes OOM intentionally to cause leaked filesystem connection in the current code base. With this patch, leakage doesn't occurs. ```scala // This should be tested manually because it raises OOM intentionally // in order to cause `Leaked filesystem connection`. test("SPARK-23399 Register a task completion listener first for OrcColumnarBatchReader") { withSQLConf(SQLConf.ORC_VECTORIZED_READER_BATCH_SIZE.key -> s"${Int.MaxValue}") { withTempDir { dir => val basePath = dir.getCanonicalPath Seq(0).toDF("a").write.format("orc").save(new Path(basePath, "first").toString) Seq(1).toDF("a").write.format("orc").save(new Path(basePath, "second").toString) val df = spark.read.orc( new Path(basePath, "first").toString, new Path(basePath, "second").toString) val e = intercept[SparkException] { df.collect() } assert(e.getCause.isInstanceOf[OutOfMemoryError]) } } } ``` Author: Dongjoon Hyun <[email protected]> Closes #20590 from dongjoon-hyun/SPARK-23399. (cherry picked from commit 357babd) Signed-off-by: Wenchen Fan <[email protected]>

[SPARK-23399][SQL] Register a task completion listner first for OrcCo…

198f186

…lumnarBatchReader

dongjoon-hyun commented Feb 12, 2018

View reviewed changes

dongjoon-hyun changed the title ~~[SPARK-23399][SQL] Register a task completion listner first for OrcColumnarBatchReader~~ [SPARK-23399][SQL] Register a task completion listener first for OrcColumnarBatchReader Feb 12, 2018

dongjoon-hyun mentioned this pull request Feb 12, 2018

[SPARK-23390][SQL] Flaky Test Suite: FileBasedDataSourceSuite in Spark 2.3/hadoop 2.7 #20584

Closed

Add a manual test case.

3b8cb0a

dongjoon-hyun commented Feb 13, 2018

View reviewed changes

fix typo

d4cc32e

cloud-fan reviewed Feb 13, 2018

View reviewed changes

Remove test case.

5a9fa0b

kiszk reviewed Feb 13, 2018

View reviewed changes

Add comments.

18c2485

asfgit closed this in 357babd Feb 14, 2018

dongjoon-hyun deleted the SPARK-23399 branch February 14, 2018 16:28

[SPARK-23399][SQL] Register a task completion listener first for OrcColumnarBatchReader #20590

[SPARK-23399][SQL] Register a task completion listener first for OrcColumnarBatchReader #20590

Uh oh!

Conversation

dongjoon-hyun commented Feb 12, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

dongjoon-hyun Feb 12, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Feb 12, 2018

Choose a reason for hiding this comment

Uh oh!

viirya Feb 13, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya Feb 13, 2018

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Feb 13, 2018

Uh oh!

kiszk commented Feb 13, 2018

Uh oh!

kiszk commented Feb 13, 2018

Uh oh!

cloud-fan commented Feb 13, 2018

Uh oh!

cloud-fan commented Feb 13, 2018

Uh oh!

dongjoon-hyun commented Feb 13, 2018

Uh oh!

dongjoon-hyun Feb 13, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan Feb 13, 2018

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Feb 13, 2018

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Feb 13, 2018

Uh oh!

SparkQA commented Feb 13, 2018

Uh oh!

kiszk commented Feb 13, 2018

Uh oh!

kiszk Feb 13, 2018

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Feb 13, 2018

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Feb 13, 2018

Uh oh!

SparkQA commented Feb 13, 2018

Uh oh!

SparkQA commented Feb 13, 2018

Uh oh!

dongjoon-hyun commented Feb 13, 2018

Uh oh!

SparkQA commented Feb 14, 2018

Uh oh!

dongjoon-hyun commented Feb 14, 2018

Uh oh!

cloud-fan commented Feb 14, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

dongjoon-hyun commented Feb 12, 2018 •

edited

Loading

dongjoon-hyun Feb 12, 2018 •

edited

Loading

viirya Feb 13, 2018 •

edited

Loading

dongjoon-hyun Feb 13, 2018 •

edited

Loading