You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/07/21 09:48:13 UTC
[GitHub] [spark] LuciferYang opened a new pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
LuciferYang opened a new pull request #33460:
URL: https://github.com/apache/spark/pull/33460
### What changes were proposed in this pull request?
The main change of this pr is move `writer.close()` before `success = true` to ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method.
### Why are the changes needed?
Avoid setting `success = true` first and then failure of close spill file
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Pass the Jenkins or GitHub Action
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r675279952
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
Review comment:
There is another `Utils` in `org.apache.spark.util.collection` package, so `util.Utils` is used here.
a57e76a change to import `Utils` directly and rename it to `UUtils`
Do you have any other suggestions for the naming of `UUtils`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884806782
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46001/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885420116
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141530/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r674612080
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
+ spillFilesCreated.clear()
+
+ val env: SparkEnv = mock(classOf[SparkEnv])
+ SparkEnv.set(env)
+
+ conf = new SparkConf()
+ when(SparkEnv.get.conf).thenReturn(conf)
+
+ val serializer = new KryoSerializer(conf)
+ when(SparkEnv.get.serializer).thenReturn(serializer)
+
+ blockManager = mock(classOf[BlockManager])
+ when(SparkEnv.get.blockManager).thenReturn(blockManager)
+
+ val manager = new SerializerManager(serializer, conf)
+ when(blockManager.serializerManager).thenReturn(manager)
+
+ diskBlockManager = mock(classOf[DiskBlockManager])
+ when(blockManager.diskBlockManager).thenReturn(diskBlockManager)
+
+ taskContext = mock(classOf[TaskContext])
+ val memoryManager = new TestMemoryManager(conf)
+ taskMemoryManager = new TaskMemoryManager(memoryManager, 0)
+ when(taskContext.taskMemoryManager()).thenReturn(taskMemoryManager)
+
+ when(diskBlockManager.createTempShuffleBlock())
+ .thenAnswer((_: InvocationOnMock) => {
+ import org.apache.spark.storage.TempShuffleBlockId
+ import java.util.UUID
+ val blockId = TempShuffleBlockId(UUID.randomUUID)
+ val file = File.createTempFile("spillFile", ".spill", tempDir)
+ spillFilesCreated += file
+ (blockId, file)
+ })
+ }
+
+ override protected def afterEach(): Unit = {
+ util.Utils.deleteRecursively(tempDir)
+ SparkEnv.set(null)
+
+ val leakedMemory = taskMemoryManager.cleanUpAllAllocatedMemory
+ if (leakedMemory != 0) {
+ fail("Test leaked " + leakedMemory + " bytes of managed memory")
+ }
+ }
+
+ test("SPARK-36242 Spill File should not exists if writer close fails") {
+ // Prepare the data and ensure that the amount of data let the `spill()` method
+ // to enter the `objectsWritten > 0` branch
+ val writeSize = conf.get(config.SHUFFLE_SPILL_BATCH_SIZE) + 1
+ val dataBuffer = new PartitionedPairBuffer[Int, Int]
+ (0 until writeSize.toInt).foreach(i => dataBuffer.insert(0, 0, i))
+
+ val externalSorter = new TestExternalSorter[Int, Int, Int](taskContext)
+
+ // Mock the answer of `blockManager.getDiskWriter` and let the `close()` method of
+ // `DiskBlockObjectWriter` throw IOException.
+ val errorMessage = "Spill file close failed"
+ when(blockManager.getDiskWriter(
+ any(classOf[BlockId]),
+ any(classOf[File]),
+ any(classOf[SerializerInstance]),
+ anyInt(),
+ any(classOf[ShuffleWriteMetrics])
+ )).thenAnswer((invocation: InvocationOnMock) => {
+ val args = invocation.getArguments
+ new DiskBlockObjectWriter(
+ args(1).asInstanceOf[File],
+ blockManager.serializerManager,
+ args(2).asInstanceOf[SerializerInstance],
+ args(3).asInstanceOf[Int],
+ false,
+ args(4).asInstanceOf[ShuffleWriteMetrics],
+ args(0).asInstanceOf[BlockId]
+ ) {
+ override def close(): Unit = throw new IOException(errorMessage)
+ }
+ })
+
+ val ioe = intercept[IOException] {
+ externalSorter.spill(dataBuffer)
+ }
+
+ ioe.getMessage.equals(errorMessage)
+ // The `TempShuffleBlock` create by diskBlockManager
+ // will remain before SPARK-36242
+ assert(!spillFilesCreated(0).exists())
Review comment:
Let me check this case with `Scala 2.13`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884059594
**[Test build #141402 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141402/testReport)** for PR 33460 at commit [`1e970f3`](https://github.com/apache/spark/commit/1e970f3f2cf44adafba58e45de2309975721f5c8).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884100746
**[Test build #141409 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141409/testReport)** for PR 33460 at commit [`f38d735`](https://github.com/apache/spark/commit/f38d735b39fb7b20532ce8da4204bd5bb88a49d8).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r674608769
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
+ spillFilesCreated.clear()
+
+ val env: SparkEnv = mock(classOf[SparkEnv])
+ SparkEnv.set(env)
+
+ conf = new SparkConf()
+ when(SparkEnv.get.conf).thenReturn(conf)
+
+ val serializer = new KryoSerializer(conf)
+ when(SparkEnv.get.serializer).thenReturn(serializer)
+
+ blockManager = mock(classOf[BlockManager])
+ when(SparkEnv.get.blockManager).thenReturn(blockManager)
+
+ val manager = new SerializerManager(serializer, conf)
+ when(blockManager.serializerManager).thenReturn(manager)
+
+ diskBlockManager = mock(classOf[DiskBlockManager])
+ when(blockManager.diskBlockManager).thenReturn(diskBlockManager)
+
+ taskContext = mock(classOf[TaskContext])
+ val memoryManager = new TestMemoryManager(conf)
+ taskMemoryManager = new TaskMemoryManager(memoryManager, 0)
+ when(taskContext.taskMemoryManager()).thenReturn(taskMemoryManager)
+
+ when(diskBlockManager.createTempShuffleBlock())
+ .thenAnswer((_: InvocationOnMock) => {
+ import org.apache.spark.storage.TempShuffleBlockId
+ import java.util.UUID
+ val blockId = TempShuffleBlockId(UUID.randomUUID)
+ val file = File.createTempFile("spillFile", ".spill", tempDir)
+ spillFilesCreated += file
+ (blockId, file)
+ })
+ }
+
+ override protected def afterEach(): Unit = {
+ util.Utils.deleteRecursively(tempDir)
+ SparkEnv.set(null)
+
+ val leakedMemory = taskMemoryManager.cleanUpAllAllocatedMemory
+ if (leakedMemory != 0) {
+ fail("Test leaked " + leakedMemory + " bytes of managed memory")
+ }
+ }
+
+ test("SPARK-36242 Spill File should not exists if writer close fails") {
+ // Prepare the data and ensure that the amount of data let the `spill()` method
+ // to enter the `objectsWritten > 0` branch
+ val writeSize = conf.get(config.SHUFFLE_SPILL_BATCH_SIZE) + 1
+ val dataBuffer = new PartitionedPairBuffer[Int, Int]
+ (0 until writeSize.toInt).foreach(i => dataBuffer.insert(0, 0, i))
+
+ val externalSorter = new TestExternalSorter[Int, Int, Int](taskContext)
+
+ // Mock the answer of `blockManager.getDiskWriter` and let the `close()` method of
+ // `DiskBlockObjectWriter` throw IOException.
+ val errorMessage = "Spill file close failed"
+ when(blockManager.getDiskWriter(
+ any(classOf[BlockId]),
+ any(classOf[File]),
+ any(classOf[SerializerInstance]),
+ anyInt(),
+ any(classOf[ShuffleWriteMetrics])
+ )).thenAnswer((invocation: InvocationOnMock) => {
+ val args = invocation.getArguments
+ new DiskBlockObjectWriter(
+ args(1).asInstanceOf[File],
+ blockManager.serializerManager,
+ args(2).asInstanceOf[SerializerInstance],
+ args(3).asInstanceOf[Int],
+ false,
+ args(4).asInstanceOf[ShuffleWriteMetrics],
+ args(0).asInstanceOf[BlockId]
+ ) {
+ override def close(): Unit = throw new IOException(errorMessage)
+ }
+ })
+
+ val ioe = intercept[IOException] {
+ externalSorter.spill(dataBuffer)
+ }
+
+ ioe.getMessage.equals(errorMessage)
+ // The `TempShuffleBlock` create by diskBlockManager
+ // will remain before SPARK-36242
+ assert(!spillFilesCreated(0).exists())
Review comment:
@mridulm @Ngone51 @HyukjinKwon before this pr, `assert(!spillFilesCreated(0).exists())` will failed, I tested it manually.
```
ExternalSorterSpillSuite.this.spillFilesCreated.apply(0).exists() was true
ScalaTestFailureLocation: org.apache.spark.util.collection.ExternalSorterSpillSuite at (ExternalSorterSpillSuite.scala:137)
org.scalatest.exceptions.TestFailedException: ExternalSorterSpillSuite.this.spillFilesCreated.apply(0).exists() was true
at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
at org.scalatest.Assertions$.newAssertionFailedException(Assertions.scala:1231)
at org.scalatest.Assertions$AssertionsHelper.macroAssert(Assertions.scala:1295)
at org.apache.spark.util.collection.ExternalSorterSpillSuite.$anonfun$new$1(ExternalSorterSpillSuite.scala:137)
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884761014
**[Test build #141483 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141483/testReport)** for PR 33460 at commit [`65e6987`](https://github.com/apache/spark/commit/65e6987d9dc9a2f9b26bf76e195b775b6dda20f3).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-886328947
> BTW, this looks like a bug fix. Could you make a backport to branch-3.1/3.0 please?
@dongjoon-hyun OK, I'll do it as soon as possible
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884900205
**[Test build #141483 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141483/testReport)** for PR 33460 at commit [`65e6987`](https://github.com/apache/spark/commit/65e6987d9dc9a2f9b26bf76e195b775b6dda20f3).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884663892
> Mock blockManager.getDiskWriter ? See UnsafeShuffleWriterSuite for an example
Thx ~
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
mridulm commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884651790
Mock `blockManager.getDiskWriter` ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885367259
**[Test build #141530 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141530/testReport)** for PR 33460 at commit [`a57e76a`](https://github.com/apache/spark/commit/a57e76a951cd87b8b0b86fbb5fa9a95e5ee273e6).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885845413
+1, LGTM. Thank you, @LuciferYang and all!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884257270
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141409/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884124449
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45920/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 closed pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
Ngone51 closed pull request #33460:
URL: https://github.com/apache/spark/pull/33460
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885417668
**[Test build #141530 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141530/testReport)** for PR 33460 at commit [`a57e76a`](https://github.com/apache/spark/commit/a57e76a951cd87b8b0b86fbb5fa9a95e5ee273e6).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r675279952
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
Review comment:
There is another `Utils` in `org.apache.spark.util.collection` package, so `util.Utils` is used here.
a57e76a change to import `Utils` directly and rename it to `UUtils`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885846967
BTW, this looks like a bug fix. Could you make a backport to branch-3.1/3.0 please?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885713417
Thanks, merged to master/3.2.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884168403
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884761014
**[Test build #141483 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141483/testReport)** for PR 33460 at commit [`65e6987`](https://github.com/apache/spark/commit/65e6987d9dc9a2f9b26bf76e195b775b6dda20f3).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884902585
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141483/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884829923
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46001/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang edited a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang edited a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884642011
> I would prefer if we had a test for this btw @LuciferYang.
How can I create a successful `flush` but fail to `writer.close()` case when calling this method?
`writer` is a local variable in `ExternalSorter#spillMemoryIteratorToDisk` method and `spillMemoryIteratorToDisk` is `private` method, so it's hard to mock writer`'s behavior.
Do you have any suggestions for this? @mridulm @HyukjinKwon @Ngone51
Or do we need more refactoring work to make this method easy to test?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885401006
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46048/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884068093
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141402/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r675404536
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
Review comment:
ok~
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r675277921
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
+ spillFilesCreated.clear()
+
+ val env: SparkEnv = mock(classOf[SparkEnv])
+ SparkEnv.set(env)
+
+ conf = new SparkConf()
+ when(SparkEnv.get.conf).thenReturn(conf)
+
+ val serializer = new KryoSerializer(conf)
+ when(SparkEnv.get.serializer).thenReturn(serializer)
+
+ blockManager = mock(classOf[BlockManager])
+ when(SparkEnv.get.blockManager).thenReturn(blockManager)
+
+ val manager = new SerializerManager(serializer, conf)
+ when(blockManager.serializerManager).thenReturn(manager)
+
+ diskBlockManager = mock(classOf[DiskBlockManager])
+ when(blockManager.diskBlockManager).thenReturn(diskBlockManager)
+
+ taskContext = mock(classOf[TaskContext])
+ val memoryManager = new TestMemoryManager(conf)
+ taskMemoryManager = new TaskMemoryManager(memoryManager, 0)
+ when(taskContext.taskMemoryManager()).thenReturn(taskMemoryManager)
+
+ when(diskBlockManager.createTempShuffleBlock())
+ .thenAnswer((_: InvocationOnMock) => {
+ import org.apache.spark.storage.TempShuffleBlockId
+ import java.util.UUID
Review comment:
ok ~
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884168403
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884257270
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141409/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884641919
> I would prefer if we had a test for this btw @LuciferYang.
How can I create a successful `flush` but fail to `writer.close()` case when calling this method?
`writer` is a local variable in `ExternalSorter#spillMemoryIteratorToDisk` method and `spillMemoryIteratorToDisk` is `private` method, so it's hard to mock writer`'s behavior.
Do you have any suggestions for this? @mridulm @HyukjinKwon @Ngone51
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884067928
**[Test build #141402 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141402/testReport)** for PR 33460 at commit [`1e970f3`](https://github.com/apache/spark/commit/1e970f3f2cf44adafba58e45de2309975721f5c8).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r674623176
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
+ spillFilesCreated.clear()
+
+ val env: SparkEnv = mock(classOf[SparkEnv])
+ SparkEnv.set(env)
+
+ conf = new SparkConf()
+ when(SparkEnv.get.conf).thenReturn(conf)
+
+ val serializer = new KryoSerializer(conf)
+ when(SparkEnv.get.serializer).thenReturn(serializer)
+
+ blockManager = mock(classOf[BlockManager])
+ when(SparkEnv.get.blockManager).thenReturn(blockManager)
+
+ val manager = new SerializerManager(serializer, conf)
+ when(blockManager.serializerManager).thenReturn(manager)
+
+ diskBlockManager = mock(classOf[DiskBlockManager])
+ when(blockManager.diskBlockManager).thenReturn(diskBlockManager)
+
+ taskContext = mock(classOf[TaskContext])
+ val memoryManager = new TestMemoryManager(conf)
+ taskMemoryManager = new TaskMemoryManager(memoryManager, 0)
+ when(taskContext.taskMemoryManager()).thenReturn(taskMemoryManager)
+
+ when(diskBlockManager.createTempShuffleBlock())
+ .thenAnswer((_: InvocationOnMock) => {
+ import org.apache.spark.storage.TempShuffleBlockId
+ import java.util.UUID
+ val blockId = TempShuffleBlockId(UUID.randomUUID)
+ val file = File.createTempFile("spillFile", ".spill", tempDir)
+ spillFilesCreated += file
+ (blockId, file)
+ })
+ }
+
+ override protected def afterEach(): Unit = {
+ util.Utils.deleteRecursively(tempDir)
+ SparkEnv.set(null)
+
+ val leakedMemory = taskMemoryManager.cleanUpAllAllocatedMemory
+ if (leakedMemory != 0) {
+ fail("Test leaked " + leakedMemory + " bytes of managed memory")
+ }
+ }
+
+ test("SPARK-36242 Spill File should not exists if writer close fails") {
+ // Prepare the data and ensure that the amount of data let the `spill()` method
+ // to enter the `objectsWritten > 0` branch
+ val writeSize = conf.get(config.SHUFFLE_SPILL_BATCH_SIZE) + 1
+ val dataBuffer = new PartitionedPairBuffer[Int, Int]
+ (0 until writeSize.toInt).foreach(i => dataBuffer.insert(0, 0, i))
+
+ val externalSorter = new TestExternalSorter[Int, Int, Int](taskContext)
+
+ // Mock the answer of `blockManager.getDiskWriter` and let the `close()` method of
+ // `DiskBlockObjectWriter` throw IOException.
+ val errorMessage = "Spill file close failed"
+ when(blockManager.getDiskWriter(
+ any(classOf[BlockId]),
+ any(classOf[File]),
+ any(classOf[SerializerInstance]),
+ anyInt(),
+ any(classOf[ShuffleWriteMetrics])
+ )).thenAnswer((invocation: InvocationOnMock) => {
+ val args = invocation.getArguments
+ new DiskBlockObjectWriter(
+ args(1).asInstanceOf[File],
+ blockManager.serializerManager,
+ args(2).asInstanceOf[SerializerInstance],
+ args(3).asInstanceOf[Int],
+ false,
+ args(4).asInstanceOf[ShuffleWriteMetrics],
+ args(0).asInstanceOf[BlockId]
+ ) {
+ override def close(): Unit = throw new IOException(errorMessage)
+ }
+ })
+
+ val ioe = intercept[IOException] {
+ externalSorter.spill(dataBuffer)
+ }
+
+ ioe.getMessage.equals(errorMessage)
+ // The `TempShuffleBlock` create by diskBlockManager
+ // will remain before SPARK-36242
+ assert(!spillFilesCreated(0).exists())
Review comment:
> Let me check this case with Scala 2.13
Manual test passed
`ExternalSorterSuite` relies on 'LocalSparkContext', so I added a new test file
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884902585
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141483/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-886345018
@dongjoon-hyun
branch-3.1: https://github.com/apache/spark/pull/33513
branch-3.0: https://github.com/apache/spark/pull/33514
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884162529
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45926/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884829923
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46001/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm edited a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
mridulm edited a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884651790
Mock `blockManager.getDiskWriter` ? See `UnsafeShuffleWriterSuite` for an example
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884642011
> I would prefer if we had a test for this btw @LuciferYang.
How can I create a successful `flush` but fail to `writer.close()` case when calling this method?
`writer` is a local variable in `ExternalSorter#spillMemoryIteratorToDisk` method and `spillMemoryIteratorToDisk` is `private` method, so it's hard to mock writer`'s behavior.
Do you have any suggestions for this? @mridulm @HyukjinKwon @Ngone51
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884829897
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46001/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r675309400
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
Review comment:
I see. Looks fine either way.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885367259
**[Test build #141530 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141530/testReport)** for PR 33460 at commit [`a57e76a`](https://github.com/apache/spark/commit/a57e76a951cd87b8b0b86fbb5fa9a95e5ee273e6).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r675279756
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
+ spillFilesCreated.clear()
+
+ val env: SparkEnv = mock(classOf[SparkEnv])
+ SparkEnv.set(env)
+
+ conf = new SparkConf()
+ when(SparkEnv.get.conf).thenReturn(conf)
+
+ val serializer = new KryoSerializer(conf)
+ when(SparkEnv.get.serializer).thenReturn(serializer)
+
+ blockManager = mock(classOf[BlockManager])
+ when(SparkEnv.get.blockManager).thenReturn(blockManager)
+
+ val manager = new SerializerManager(serializer, conf)
+ when(blockManager.serializerManager).thenReturn(manager)
+
+ diskBlockManager = mock(classOf[DiskBlockManager])
+ when(blockManager.diskBlockManager).thenReturn(diskBlockManager)
+
+ taskContext = mock(classOf[TaskContext])
+ val memoryManager = new TestMemoryManager(conf)
+ taskMemoryManager = new TaskMemoryManager(memoryManager, 0)
+ when(taskContext.taskMemoryManager()).thenReturn(taskMemoryManager)
+
+ when(diskBlockManager.createTempShuffleBlock())
+ .thenAnswer((_: InvocationOnMock) => {
+ import org.apache.spark.storage.TempShuffleBlockId
+ import java.util.UUID
Review comment:
done
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r675274673
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
Review comment:
import `Utils` directly?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884068093
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141402/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r675274774
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
+ spillFilesCreated.clear()
+
+ val env: SparkEnv = mock(classOf[SparkEnv])
+ SparkEnv.set(env)
+
+ conf = new SparkConf()
+ when(SparkEnv.get.conf).thenReturn(conf)
+
+ val serializer = new KryoSerializer(conf)
+ when(SparkEnv.get.serializer).thenReturn(serializer)
+
+ blockManager = mock(classOf[BlockManager])
+ when(SparkEnv.get.blockManager).thenReturn(blockManager)
+
+ val manager = new SerializerManager(serializer, conf)
+ when(blockManager.serializerManager).thenReturn(manager)
+
+ diskBlockManager = mock(classOf[DiskBlockManager])
+ when(blockManager.diskBlockManager).thenReturn(diskBlockManager)
+
+ taskContext = mock(classOf[TaskContext])
+ val memoryManager = new TestMemoryManager(conf)
+ taskMemoryManager = new TaskMemoryManager(memoryManager, 0)
+ when(taskContext.taskMemoryManager()).thenReturn(taskMemoryManager)
+
+ when(diskBlockManager.createTempShuffleBlock())
+ .thenAnswer((_: InvocationOnMock) => {
+ import org.apache.spark.storage.TempShuffleBlockId
+ import java.util.UUID
Review comment:
Move to the imports group?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885381684
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46048/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884641919
> I would prefer if we had a test for this btw @LuciferYang.
How can I create a successful `flush` but fail to `writer.close()` case when calling this method?
`writer` is a local variable in `ExternalSorter#spillMemoryIteratorToDisk` method and `spillMemoryIteratorToDisk` is `private` method, so it's hard to mock writer`'s behavior.
Do you have any suggestions for this? @mridulm @HyukjinKwon @Ngone51
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884100746
**[Test build #141409 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141409/testReport)** for PR 33460 at commit [`f38d735`](https://github.com/apache/spark/commit/f38d735b39fb7b20532ce8da4204bd5bb88a49d8).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884149230
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45920/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885392575
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46048/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885401006
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/46048/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884245303
**[Test build #141409 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141409/testReport)** for PR 33460 at commit [`f38d735`](https://github.com/apache/spark/commit/f38d735b39fb7b20532ce8da4204bd5bb88a49d8).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-884059594
**[Test build #141402 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141402/testReport)** for PR 33460 at commit [`1e970f3`](https://github.com/apache/spark/commit/1e970f3f2cf44adafba58e45de2309975721f5c8).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] LuciferYang commented on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-886347233
thx ~ @Ngone51 @mridulm @dongjoon-hyun
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33460:
URL: https://github.com/apache/spark/pull/33460#issuecomment-885420116
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141530/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #33460: [SPARK-36242][CORE] Ensure spill file closed before set `success = true` in `ExternalSorter.spillMemoryIteratorToDisk` method
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #33460:
URL: https://github.com/apache/spark/pull/33460#discussion_r675274990
##########
File path: core/src/test/scala/org/apache/spark/util/collection/ExternalSorterSpillSuite.scala
##########
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.collection
+
+import java.io.{File, IOException}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.mockito.ArgumentMatchers.{any, anyInt}
+import org.mockito.Mockito.{mock, when}
+import org.mockito.invocation.InvocationOnMock
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite, TaskContext}
+import org.apache.spark.executor.ShuffleWriteMetrics
+import org.apache.spark.internal.config
+import org.apache.spark.memory.{TaskMemoryManager, TestMemoryManager}
+import org.apache.spark.serializer.{KryoSerializer, SerializerInstance, SerializerManager}
+import org.apache.spark.storage.{BlockId, BlockManager, DiskBlockManager, DiskBlockObjectWriter}
+import org.apache.spark.util
+
+class ExternalSorterSpillSuite extends SparkFunSuite with BeforeAndAfterEach {
+
+ private val spillFilesCreated = ArrayBuffer.empty[File]
+
+ private var tempDir: File = _
+ private var conf: SparkConf = _
+ private var taskMemoryManager: TaskMemoryManager = _
+
+ private var blockManager: BlockManager = _
+ private var diskBlockManager: DiskBlockManager = _
+ private var taskContext: TaskContext = _
+
+ override protected def beforeEach(): Unit = {
+ tempDir = util.Utils.createTempDir(null, "test")
+ spillFilesCreated.clear()
+
+ val env: SparkEnv = mock(classOf[SparkEnv])
+ SparkEnv.set(env)
+
+ conf = new SparkConf()
+ when(SparkEnv.get.conf).thenReturn(conf)
+
+ val serializer = new KryoSerializer(conf)
+ when(SparkEnv.get.serializer).thenReturn(serializer)
+
+ blockManager = mock(classOf[BlockManager])
+ when(SparkEnv.get.blockManager).thenReturn(blockManager)
+
+ val manager = new SerializerManager(serializer, conf)
+ when(blockManager.serializerManager).thenReturn(manager)
+
+ diskBlockManager = mock(classOf[DiskBlockManager])
+ when(blockManager.diskBlockManager).thenReturn(diskBlockManager)
+
+ taskContext = mock(classOf[TaskContext])
+ val memoryManager = new TestMemoryManager(conf)
+ taskMemoryManager = new TaskMemoryManager(memoryManager, 0)
+ when(taskContext.taskMemoryManager()).thenReturn(taskMemoryManager)
+
+ when(diskBlockManager.createTempShuffleBlock())
+ .thenAnswer((_: InvocationOnMock) => {
+ import org.apache.spark.storage.TempShuffleBlockId
+ import java.util.UUID
+ val blockId = TempShuffleBlockId(UUID.randomUUID)
+ val file = File.createTempFile("spillFile", ".spill", tempDir)
+ spillFilesCreated += file
+ (blockId, file)
+ })
+ }
+
+ override protected def afterEach(): Unit = {
+ util.Utils.deleteRecursively(tempDir)
+ SparkEnv.set(null)
+
+ val leakedMemory = taskMemoryManager.cleanUpAllAllocatedMemory
+ if (leakedMemory != 0) {
+ fail("Test leaked " + leakedMemory + " bytes of managed memory")
+ }
+ }
+
+ test("SPARK-36242 Spill File should not exists if writer close fails") {
+ // Prepare the data and ensure that the amount of data let the `spill()` method
+ // to enter the `objectsWritten > 0` branch
+ val writeSize = conf.get(config.SHUFFLE_SPILL_BATCH_SIZE) + 1
+ val dataBuffer = new PartitionedPairBuffer[Int, Int]
+ (0 until writeSize.toInt).foreach(i => dataBuffer.insert(0, 0, i))
+
+ val externalSorter = new TestExternalSorter[Int, Int, Int](taskContext)
+
+ // Mock the answer of `blockManager.getDiskWriter` and let the `close()` method of
+ // `DiskBlockObjectWriter` throw IOException.
+ val errorMessage = "Spill file close failed"
+ when(blockManager.getDiskWriter(
+ any(classOf[BlockId]),
+ any(classOf[File]),
+ any(classOf[SerializerInstance]),
+ anyInt(),
+ any(classOf[ShuffleWriteMetrics])
+ )).thenAnswer((invocation: InvocationOnMock) => {
+ val args = invocation.getArguments
+ new DiskBlockObjectWriter(
+ args(1).asInstanceOf[File],
+ blockManager.serializerManager,
+ args(2).asInstanceOf[SerializerInstance],
+ args(3).asInstanceOf[Int],
+ false,
+ args(4).asInstanceOf[ShuffleWriteMetrics],
+ args(0).asInstanceOf[BlockId]
+ ) {
+ override def close(): Unit = throw new IOException(errorMessage)
+ }
+ })
+
+ val ioe = intercept[IOException] {
+ externalSorter.spill(dataBuffer)
+ }
+
+ ioe.getMessage.equals(errorMessage)
+ // The `TempShuffleBlock` create by diskBlockManager
+ // will remain before SPARK-36242
+ assert(!spillFilesCreated(0).exists())
Review comment:
Thanks for the great work!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org