You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "anishshri-db (via GitHub)" <gi...@apache.org> on 2023/08/31 02:36:42 UTC
[GitHub] [spark] anishshri-db opened a new pull request, #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
anishshri-db opened a new pull request, #42742:
URL: https://github.com/apache/spark/pull/42742
### What changes were proposed in this pull request?
Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
### Why are the changes needed?
Currently the `putIteratorAsBytes` can remain hung even if thread interrupt is received on task cancellation leading to the task reaper killing the executor JVM eventually. The reason for this is that the interrupt is never processed within the while loop for the unroll block which leads to the task continuing running beyond the reaper timeout.
Attached the logs here for a particular task/thread:
```
10.1.121.105/app-20230824190614-0000/735/stderr.txt:55427:23/08/29 12:01:51 INFO CoarseGrainedExecutorBackend: Got assigned task 222564684
10.1.121.105/app-20230824190614-0000/735/stderr.txt:55494:23/08/29 12:01:51 INFO Executor: Running task 6.0 in stage 900216.0 (TID 222564684)
10.1.121.105/app-20230824190614-0000/735/stderr.txt:55983:23/08/29 12:03:22 INFO Executor: Executor is trying to kill task 6.0 in stage 900216.0 (TID 222564684), reason: another attempt succeeded
```
```
10.1.121.105/app-20230824190614-0000/735/stderr.txt:55987:23/08/29 12:03:22 INFO ShuffleMapTask: Trying to kill task 6.0 in stage 900216.0 (TID 222564684) with reason=another attempt succeeded and current stackTrace:
net.jpountz.lz4.LZ4JNI.LZ4_compress_limitedOutput(Native Method)
at net.jpountz.lz4.LZ4JNICompressor.compress(LZ4JNICompressor.java:36)
at net.jpountz.lz4.LZ4Compressor.compress(LZ4Compressor.java:95)
at net.jpountz.lz4.LZ4BlockOutputStream.flushBufferedData(LZ4BlockOutputStream.java:208)
at net.jpountz.lz4.LZ4BlockOutputStream.write(LZ4BlockOutputStream.java:176)
at java.io.ObjectOutputStream$BlockDataOutputStream.drain(ObjectOutputStream.java:1877)
at java.io.ObjectOutputStream$BlockDataOutputStream.setBlockDataMode(ObjectOutputStream.java:1786)
at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1460)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:44)
at org.apache.spark.storage.memory.SerializedValuesHolder.storeValue(MemoryStore.scala:728)
at org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:224)
at org.apache.spark.storage.memory.MemoryStore.putIteratorAsBytes(MemoryStore.scala:352)
at org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1447)
at org.apache.spark.storage.BlockManager$$Lambda$732/1315363341.apply(Unknown Source)
at org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1357)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1421)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1240)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:391)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:342)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
at org.apache.spark.scheduler.ShuffleMapTask.$anonfun$runTask$3(ShuffleMapTask.scala:81)
at org.apache.spark.scheduler.ShuffleMapTask$$Lambda$1195/1841467537.apply(Unknown Source)
at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at org.apache.spark.scheduler.ShuffleMapTask.$anonfun$runTask$1(ShuffleMapTask.scala:81)
at org.apache.spark.scheduler.ShuffleMapTask$$Lambda$1050/445760861.apply(Unknown Source)
at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
at org.apache.spark.scheduler.Task.doRunTask(Task.scala:153)
at org.apache.spark.scheduler.Task.$anonfun$run$1(Task.scala:122)
at org.apache.spark.scheduler.Task$$Lambda$925/288368281.apply(Unknown Source)
at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at org.apache.spark.scheduler.Task.run(Task.scala:94)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$13(Executor.scala:819)
at org.apache.spark.executor.Executor$TaskRunner$$Lambda$909/144086205.apply(Unknown Source)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1657)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:822)
at org.apache.spark.executor.Executor$TaskRunner$$Lambda$793/213482103.apply$mcV$sp(Unknown Source)
at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
at com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:678)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
...
10.1.121.105/app-20230824190614-0000/735/stderr.txt:58310:23/08/29 12:04:31 WARN Executor: Killed task 222564684 is still running after 68908 ms
10.1.121.105/app-20230824190614-0000/735/stderr.txt:58361:23/08/29 12:04:31 WARN Executor: Thread dump from task 222564684:
java.lang.System.identityHashCode(Native Method)java.io.ObjectOutputStream$HandleTable.hash(ObjectOutputStream.java:2360)java.io.ObjectOutputStream$HandleTable.lookup(ObjectOutputStream.java:2293)java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1116)java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:44)org.apache.spark.storage.memory.SerializedValuesHolder.storeValue(MemoryStore.scala:728)org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:224)org.apache.spark.storage.memory.MemoryStore.putIteratorAsBytes(MemoryStore.scala:352)org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1447)org.apache.spark.storage.BlockManager$$Lambda$732/1315363341.apply(Unknown Source)org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1357)org.apache.spark.storage.BlockManager.doPutItera
tor(BlockManager.scala:1421)org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1240)org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:391)org.apache.spark.rdd.RDD.iterator(RDD.scala:342)org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)org.apache.spark.rdd.RDD.iterator(RDD.scala:344)org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)org.apache.spark.rdd.RDD.iterator(RDD.scala:344)org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)org.apache.spark.rdd.RDD.iterator(RDD.scala:344)org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)org.apache.spark.scheduler.ShuffleMapTask.$anonfun$runTask$3(ShuffleMapTask.scala:81)org.apache.spark.scheduler.ShuffleMapTask$$Lambda$1195/1841467537.apply(U
nknown Source)com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)org.apache.spark.scheduler.ShuffleMapTask.$anonfun$runTask$1(ShuffleMapTask.scala:81)org.apache.spark.scheduler.ShuffleMapTask$$Lambda$1050/445760861.apply(Unknown Source)com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)org.apache.spark.scheduler.Task.doRunTask(Task.scala:153)org.apache.spark.scheduler.Task.$anonfun$run$1(Task.scala:122)org.apache.spark.scheduler.Task$$Lambda$925/288368281.apply(Unknown Source)com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)org.apache.spark.scheduler.Task.run(Task.scala:94)org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$13(Executor.scala:819)org.apache.spark.executor.Executor$TaskRunner$$Lambda$909/144086205.apply(U
nknown Source)org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1657)org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:822)org.apache.spark.executor.Executor$TaskRunner$$Lambda$793/213482103.apply$mcV$sp(Unknown Source)scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:678)java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)java.lang.Thread.run(Thread.java:750)
10.1.121.105/app-20230824190614-0000/735/stderr.txt:58388:org.apache.spark.SparkException: Killing executor JVM because killed task 222564684 could not be stopped within 60000 ms.
10.1.121.105/app-20230824190614-0000/735/stderr.txt:58810:java.lang.Error: org.apache.spark.SparkException: Killing executor JVM because killed task 222564684 could not be stopped within 60000 ms.
10.1.121.105/app-20230824190614-0000/735/stderr.txt:58814:Caused by: org.apache.spark.SparkException: Killing executor JVM because killed task 222564684 could not be stopped within 60000 ms.
```
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Ran MemoryStoreSuite
```
[info] MemoryStoreSuite:
[info] - reserve/release unroll memory (36 milliseconds)
[info] - safely unroll blocks (70 milliseconds)
[info] - safely unroll blocks through putIteratorAsValues (10 milliseconds)
[info] - safely unroll blocks through putIteratorAsValues off-heap (21 milliseconds)
[info] - safely unroll blocks through putIteratorAsBytes (138 milliseconds)
[info] - PartiallySerializedBlock.valuesIterator (6 milliseconds)
[info] - PartiallySerializedBlock.finishWritingToStream (5 milliseconds)
[info] - multiple unrolls by the same thread (8 milliseconds)
[info] - lazily create a big ByteBuffer to avoid OOM if it cannot be put into MemoryStore (3 milliseconds)
[info] - put a small ByteBuffer to MemoryStore (3 milliseconds)
[info] - SPARK-22083: Release all locks in evictBlocksToFreeSpace (43 milliseconds)
[info] - put user-defined objects to MemoryStore and remove (5 milliseconds)
[info] - put user-defined objects to MemoryStore and clear (4 milliseconds)
[info] Run completed in 1 second, 587 milliseconds.
[info] Total number of tests run: 13
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 13, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
### Was this patch authored or co-authored using generative AI tooling?
No
--
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 a diff in pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42742:
URL: https://github.com/apache/spark/pull/42742#discussion_r1314096518
##########
core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala:
##########
@@ -220,7 +220,8 @@ private[spark] class MemoryStore(
}
// Unroll this block safely, checking whether we have exceeded our threshold periodically
- while (values.hasNext && keepUnrolling) {
+ // and if no thread interrupts have been received.
+ while (values.hasNext && keepUnrolling && !Thread.currentThread().isInterrupted) {
Review Comment:
@anishshri-db Thread interruption is a rare occurrence when compared to number of `putIterator` calls - so we should try to minimize the impact for the common case, as long as it is a reasonable additional cost when interruption does occur (having said that, please see more below).
@JoshRosen It is way faster than I expected it to be [1] - while it is still slower than a local variable [2], the cost is really low when compared to the cost of other operations in the loop ... this should be just noise.
Given this, while in general I prefer to minimize unnecessary cost, I am fine with the change.
[1] Thanks for your comments, went digging into this - was fun !
I knew it was intrinsic, but the uncontended reads are much faster than I had expected (I seem to be misremembering some stats) - and thanks for jdk14 ref, was not aware of that change.
[2]
On my linux desktop: 2.175 +- 0.001 ns/op versus 2.424 +- 0.001 ns/op
On my mac, the difference is higher - but so is the variance ... so I am discounting that
--
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] HeartSaVioR commented on pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1709284624
As long as I see @mridulm is OK with the fix, I can help merging this.
Thanks! Merging to master.
--
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] anishshri-db commented on pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1700289385
cc - @JoshRosen , @HeartSaVioR - PTAL, 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] anishshri-db commented on pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1701750204
@JoshRosen - agreed, don't think its being handled currently anyway. If we want to add the catch block on the caller, I think we need to do it for `putIteratorAsValues` as well ? Do you think its ok to split this issue into 2 parts since I think the other change might need some more work ? For the current PR, I'll just modify the return value to throw the exception and then file a follow-up JIRA ticket for the cleanup related 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
[GitHub] [spark] anishshri-db commented on pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1708981207
Thanks @JoshRosen . @mridulm , @HeartSaVioR - would one of you be able to merge then ?
--
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] HeartSaVioR closed pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR closed pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
URL: https://github.com/apache/spark/pull/42742
--
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 a diff in pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42742:
URL: https://github.com/apache/spark/pull/42742#discussion_r1312531275
##########
core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala:
##########
@@ -220,7 +220,8 @@ private[spark] class MemoryStore(
}
// Unroll this block safely, checking whether we have exceeded our threshold periodically
- while (values.hasNext && keepUnrolling) {
+ // and if no thread interrupts have been received.
+ while (values.hasNext && keepUnrolling && !Thread.currentThread().isInterrupted) {
Review Comment:
I am slightly wary of calling `isInterrupted` for every iteration.
Given we already have the `if (elementsUnrolled ...` condition below, do we want to move it to that ?
IIRC `isInterrupted` is a couple of orders more expensive than a boolean check.
Something like:
```
var interrupted = false
while (values.hasNext && keepUnrolling && !interrupted) {
valueHolder ...
if (... ){
interrupted = Thread.currentThread().isInterrupted
...
}
```
--
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 a diff in pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42742:
URL: https://github.com/apache/spark/pull/42742#discussion_r1312531275
##########
core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala:
##########
@@ -220,7 +220,8 @@ private[spark] class MemoryStore(
}
// Unroll this block safely, checking whether we have exceeded our threshold periodically
- while (values.hasNext && keepUnrolling) {
+ // and if no thread interrupts have been received.
+ while (values.hasNext && keepUnrolling && !Thread.currentThread().isInterrupted) {
Review Comment:
I am slightly wary of calling `isInterrupted` for every iteration.
Given we already have the `if (elementsUnrolled ...` condition below, do we want to move it to that ?
IIRC `isInterrupted` is a couple of orders more expensive than a boolean check.
Something like:
```
var interrupted = false
while (values.hasNext && keepUnrolling && !interrupted) {
valuesHolder ...
if (elementsUnrolled ... ){
interrupted = Thread.currentThread().isInterrupted
...
}
elementsUnrolled += 1
}
if (interrupted) {
...
} else if (keepUnrolling) {
```
--
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] JoshRosen commented on a diff in pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on code in PR #42742:
URL: https://github.com/apache/spark/pull/42742#discussion_r1313513495
##########
core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala:
##########
@@ -220,7 +220,8 @@ private[spark] class MemoryStore(
}
// Unroll this block safely, checking whether we have exceeded our threshold periodically
- while (values.hasNext && keepUnrolling) {
+ // and if no thread interrupts have been received.
+ while (values.hasNext && keepUnrolling && !Thread.currentThread().isInterrupted) {
Review Comment:
> IIRC isInterrupted is a couple of orders more expensive than a boolean check.
In older JDKs it is implemented via an intrinsic and in newer ones (14+) it's just a volatile boolean check (see https://bugs.openjdk.org/browse/JDK-8229516), so I don't think it's too expensive. There's an old StackOverflow answer at https://stackoverflow.com/a/5158441/590203 offering a plausible explanation for why the intrinsic should be cheap.
On JDK 11 on my laptop, I ran a toy JMH benchmark and measured < 1ns per call.
I think these costs are very small in comparison to the per-element serialization costs.
Given this, I'm not too worried about performance regressions due to this change and think it's probably okay to check on every element rather than every 16th.
--
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] JoshRosen commented on pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1708974783
I'm traveling without a laptop right now, so I won't be able to merge this until next week.
--
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 #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1709302070
Thanks @HeartSaVioR !
--
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] anishshri-db commented on pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1708789388
@JoshRosen - seems like we are all in agreement ? Is it ok to merge the change ? 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] HeartSaVioR commented on pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "HeartSaVioR (via GitHub)" <gi...@apache.org>.
HeartSaVioR commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1700620454
cc. @jiangxb1987 @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] anishshri-db commented on pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1701860319
> As a compromise, what do you think about keeping this as-is (i.e. not throwing InterruptedException) and instead add a single taskContext.killTaskIfInterrupted call right before we return from putIteratorAsBytes? e.g.
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] anishshri-db commented on a diff in pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on code in PR #42742:
URL: https://github.com/apache/spark/pull/42742#discussion_r1312619455
##########
core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala:
##########
@@ -220,7 +220,8 @@ private[spark] class MemoryStore(
}
// Unroll this block safely, checking whether we have exceeded our threshold periodically
- while (values.hasNext && keepUnrolling) {
+ // and if no thread interrupts have been received.
+ while (values.hasNext && keepUnrolling && !Thread.currentThread().isInterrupted) {
Review Comment:
@mridulm - if we put this check inside the `if` condition below, it seems we will only set the `interrupted` flag the next time the memory check period expires ? Seems like the default value for the check period is set to 16. So, if the interrupt is received after the first element is unrolled, we will wait for 15 more elements and also do extra work in the interim, that we probably have to dispose later anyway ? So might be better to check within the `while` loop and exit earlier as we are doing in the PR ?
```
valuesHolder.storeValue(values.next())
if (elementsUnrolled % memoryCheckPeriod == 0) {
val currentSize = valuesHolder.estimatedSize()
// If our vector's size has exceeded the threshold, request more memory
```
--
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] JoshRosen commented on pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1701720571
> So it seems freeing the memory is not a problem, but if we return the InterruptedException, we would still be risking leaking the direct buffers, since we won't get a chance to register this task completion listener ? Do you think this is safe/handled elsewhere in the caller when the exception is received ?
If I understand correctly, I think this might be a pre-existing risk that we're making worse: there's nothing that prevented the old code from throwing arbitrary exceptions when computing the iterator elements.
I wonder whether we should aim to fix that pre-existing bug at a higher level. In `putIteratorAsBytes`, we have
https://github.com/apache/spark/blob/e72ce91250a9a2c40fd5ed55a50dbc46e4e7e46d/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala#L349-L367
I'm wondering whether we can restructure that code in order to wrap the `putIterator` call and dispose of the `valuesHolder` in case `putIterator` fails. Something along these lines (borrowing some code and comments from elsewhere in this part of Spark):
```scala
val putIteratorResult = Utils.tryWithSafeFinallyAndFailureCallbacks {
putIterator(blockId, values, classTag, memoryMode, valuesHolder)
}(catchBlock = {
// We want to close the output stream in order to free any resources associated with the
// serializer itself (such as Kryo's internal buffers). close() might cause data to be
// written, so redirect the output stream to discard that data.
valuesHolder.redirectableOutputStream.setOutputStream(ByteStreams.nullOutputStream())
valuesHolder.serializationStream.close()
valuesHolder.bbos.close()
valuesHolder.bbos.toChunkedByteBuffer.dispose()
})
putIteratorResult match {
case Right(storedSize) => Right(storedSize)
case Left(unrollMemoryUsedByThisBlock) =>
Left(new PartiallySerializedBlock(
this,
serializerManager,
blockId,
valuesHolder.serializationStream,
valuesHolder.redirectableStream,
unrollMemoryUsedByThisBlock,
memoryMode,
valuesHolder.bbos,
values,
classTag))
}
```
If the `putIterator()` call fails then the `catchBlock` will try to close the serialization stream and dispose of the block. I used `tryWithSafeFinallyAndFailureCallbacks` because there's non-trivial cleanup work taking place in the catch block and I didn't want that to suppress the original task exception.
I scoped the `try` block to exclude the `new PartiallySerializedBlock` because I wanted to avoid the possibility that two different pieces of cleanup logic (the task completion callback and the catch block) both call `toChunkedByteBuffer()`.
---
As I look further into the pre-existing code, I'm spotting a couple of other cases where it looks like we're not guaranteed to perform proper cleanup. For example, it looks like we're not guaranteed to close the serialization stream if downstream partial unrolling code fails (or at least it's not straightforwardly obvious that cleanup will happen).
To better test those cases, I think we should add some new unit test cases to `MemoryStoreSuite` to test scenarios where the iterator being stored throws exceptions at various points.
--
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 a diff in pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42742:
URL: https://github.com/apache/spark/pull/42742#discussion_r1312531275
##########
core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala:
##########
@@ -220,7 +220,8 @@ private[spark] class MemoryStore(
}
// Unroll this block safely, checking whether we have exceeded our threshold periodically
- while (values.hasNext && keepUnrolling) {
+ // and if no thread interrupts have been received.
+ while (values.hasNext && keepUnrolling && !Thread.currentThread().isInterrupted) {
Review Comment:
I am slightly wary of calling `isInterrupted` for every iteration.
Given we already have the `if (elementsUnrolled ...` condition below, do we want to move it to that ?
IIRC `isInterrupted` is a couple of orders more expensive than a boolean check.
Something like:
```
var interrupted = false
while (values.hasNext && keepUnrolling && !interrupted) {
valuesHolder ...
if (elementsUnrolled ... ){
interrupted = Thread.currentThread().isInterrupted
...
}
```
--
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 a diff in pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42742:
URL: https://github.com/apache/spark/pull/42742#discussion_r1314096518
##########
core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala:
##########
@@ -220,7 +220,8 @@ private[spark] class MemoryStore(
}
// Unroll this block safely, checking whether we have exceeded our threshold periodically
- while (values.hasNext && keepUnrolling) {
+ // and if no thread interrupts have been received.
+ while (values.hasNext && keepUnrolling && !Thread.currentThread().isInterrupted) {
Review Comment:
@anishshri-db Thread interruption is a rare occurrence when compared to number of `putIterator` calls - so we should try to minimize the impact for the common case, as long as it is a reasonable additional cost when interruption does occur (having said that, please see more below).
@JoshRosen It is way faster than I expected it to be [1] - while it is still slower than a local variable [2], the cost is really low when compared to the cost of other operations in the loop ... this should be just noise.
Given this, while in general I prefer to minimize unnecessary cost, I am fine with the change.
[1] Thanks for your comments, went digging into this - was fun !
I knew it was intrinsic, but the uncontended reads are much faster than I had expected (I seem to be misremembering some stats) - and thanks for jdk14 ref, was not aware of that change.
[2]
On my linux desktop: 2.175 +- 0.001 ns/op versus 2.424 +- 0.001 ns/op
--
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] anishshri-db commented on pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1701633031
> Instead, what do you think about replacing the Left(unrollMemoryUsedByThisBlock) with a throw new InterruptedException()? That exception will bubble up and help the task to exit sooner.
@JoshRosen - I was going to do exactly this initially :) But I was not sure whether this would be entirely safe or not, in terms of disposing the byte buffers.
It seems the caller relies on the result here to create `PartiallySerializedBlock` in case of failure
And within this, we add a task completion listener to ensure that we call dispose on the byte buffers ?
```
private lazy val unrolledBuffer: ChunkedByteBuffer = {
bbos.close()
bbos.toChunkedByteBuffer
}
// If the task does not fully consume `valuesIterator` or otherwise fails to consume or dispose of
// this PartiallySerializedBlock then we risk leaking of direct buffers, so we use a task
// completion listener here in order to ensure that `unrolled.dispose()` is called at least once.
// The dispose() method is idempotent, so it's safe to call it unconditionally.
Option(TaskContext.get()).foreach { taskContext =>
taskContext.addTaskCompletionListener[Unit] { _ =>
// When a task completes, its unroll memory will automatically be freed. Thus we do not call
// releaseUnrollMemoryForThisTask() here because we want to avoid double-freeing.
unrolledBuffer.dispose()
}
}
```
So it seems freeing the memory is not a problem, but if we return the `InterruptedException`, we would still be risking leaking the direct buffers, since we won't get a chance to register this task completion listener ? Do you think this is safe/handled elsewhere in the caller when the exception is received ?
--
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] JoshRosen commented on pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1701830082
> If we want to add the catch block on the caller, I think we need to do it for putIteratorAsValues as well ?
I don't think we need it there: `putIteratorAsValues` stores deserialized values, so there's no off-heap memory to be freed or serialization streams to be closed.
> Do you think its ok to split this issue into 2 parts since I think the other change might need some more work ? For the current PR, I'll just modify the return value to throw the exception and then file a follow-up JIRA ticket for the cleanup related work ?
As a compromise, what do you think about keeping this as-is (i.e. not throwing `InterruptedException`) and instead add a single `taskContext.killTaskIfInterrupted` call right before we return from `putIteratorAsBytes`? e.g.
```scala
val res = putIteratorResult match {
case Right(storedSize) => Right(storedSize)
case Left(unrollMemoryUsedByThisBlock) =>
Left(new PartiallySerializedBlock(
this,
serializerManager,
blockId,
valuesHolder.serializationStream,
valuesHolder.redirectableStream,
unrollMemoryUsedByThisBlock,
memoryMode,
valuesHolder.bbos,
values,
classTag))
}
Option(TaskContext.get()).foreach(_.killTaskIfInterrupted())
res
```
This approach will ensure that the bytebuffer cleanup logic is run (because the task completion callback will have been registered) but also ensures that we'll exit in a timely manner rather than trying to continue processing the rest of the task's rows.
--
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 a diff in pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42742:
URL: https://github.com/apache/spark/pull/42742#discussion_r1314096518
##########
core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala:
##########
@@ -220,7 +220,8 @@ private[spark] class MemoryStore(
}
// Unroll this block safely, checking whether we have exceeded our threshold periodically
- while (values.hasNext && keepUnrolling) {
+ // and if no thread interrupts have been received.
+ while (values.hasNext && keepUnrolling && !Thread.currentThread().isInterrupted) {
Review Comment:
@anishshri-db Thread interruption is a rare occurrence when compared to number of `putIterator` calls - so we should try to minimize the impact for the common case, as long as it is a reasonable additional cost when interruption does occur (having said that, please see more below).
@JoshRosen It is way faster than I expected it to be [1] - while it is still slower than a local variable [2], the cost is really low when compared to the cost of other operations in the loop ... this should be just noise.
Given this, while in general I prefer to minimize unnecessary cost, I am fine with the change.
[1] Thanks for your comments, went digging into this - was fun !
I knew it was intrinsic, but the uncontended reads are much faster than I had expected (I seem to be misremembering some stats) - and thanks for jdk14 ref, was not aware of that change.
My comment "IIRC isInterrupted is a couple of orders more expensive than a boolean check." is definitely incorrect !
[2]
On my linux desktop: 2.175 +- 0.001 ns/op versus 2.424 +- 0.001 ns/op
On my mac, the difference is higher - but so is the variance ... so I am discounting that
--
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 a diff in pull request #42742: [SPARK-45025][CORE] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #42742:
URL: https://github.com/apache/spark/pull/42742#discussion_r1314096518
##########
core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala:
##########
@@ -220,7 +220,8 @@ private[spark] class MemoryStore(
}
// Unroll this block safely, checking whether we have exceeded our threshold periodically
- while (values.hasNext && keepUnrolling) {
+ // and if no thread interrupts have been received.
+ while (values.hasNext && keepUnrolling && !Thread.currentThread().isInterrupted) {
Review Comment:
@anishshri-db Thread interruption is a rare occurrence when compared to number of `putIterator` calls - so we should try to minimize the impact for the common case, as long as it is a reasonable additional cost when interruption does occur (having said that, please see more below).
@JoshRosen It is way faster than I expected it to be [1] - while it is still slower than a local variable [2], the cost is really low when compared to the cost of other operations in the loop ... this should be just noise.
Given this, while in general I prefer to minimize unnecessary cost, I am fine with the change.
[1] Thanks for your comments, went digging into this - was fun !
I knew it was intrinsic, but the uncontended reads are much faster than I had expected (I seem to be misremembering some stats) - and thanks for jdk14 ref, was not aware of that change.
[2]
On my linux desktop: 2.175 +- 0.001 ns/op versus 2.424 +- 0.001 ns/op
On my mac, the difference is higher - but so is the variance ... so less reliable
--
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] anishshri-db commented on pull request #42742: [SPARK-45025] Allow block manager memory store iterator to handle thread interrupt and perform task completion gracefully
Posted by "anishshri-db (via GitHub)" <gi...@apache.org>.
anishshri-db commented on PR #42742:
URL: https://github.com/apache/spark/pull/42742#issuecomment-1703018412
@JoshRosen - any more comments here ? or do you think we can merge this ?
--
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