[jira] [Updated] (SPARK-43242) shuffle diagnoseCorruption should not throw Unexpected type of BlockId for ShuffleBlockBatchId
[
https://issues.apache.org/jira/browse/SPARK-43242?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zhang Liang updated SPARK-43242:
Description:
Some of our spark app throw "Unexpected type of BlockId" exception as shown
below
According to BlockId.scala, we can found format such as
*shuffle_12_5868_518_523* is type of `ShuffleBlockBatchId`, which is not
handled properly in `ShuffleBlockFetcherIterator.diagnoseCorruption`.
Moreover, the new exception thrown in `diagnose` swallow the real exception in
certain cases. Since diagnoseCorruption is always used in exception handling as
a side dish, I think it shouldn't throw exception at all
{code:java}
23/03/07 03:01:24,485 [task-result-getter-1] WARN TaskSetManager: Lost task
104.0 in stage 36.0 (TID 6169): java.lang.IllegalArgumentException: Unexpected
type of BlockId, shuffle_12_5868_518_523 at
org.apache.spark.storage.ShuffleBlockFetcherIterator.diagnoseCorruption(ShuffleBlockFetcherIterator.scala:1079)at
org.apache.spark.storage.BufferReleasingInputStream.$anonfun$tryOrFetchFailedException$1(ShuffleBlockFetcherIterator.scala:1314)
at scala.Option.map(Option.scala:230)at
org.apache.spark.storage.BufferReleasingInputStream.tryOrFetchFailedException(ShuffleBlockFetcherIterator.scala:1313)
at
org.apache.spark.storage.BufferReleasingInputStream.read(ShuffleBlockFetcherIterator.scala:1299)
at java.io.BufferedInputStream.fill(BufferedInputStream.java:246) at
java.io.BufferedInputStream.read1(BufferedInputStream.java:286) at
java.io.BufferedInputStream.read(BufferedInputStream.java:345) at
java.io.DataInputStream.read(DataInputStream.java:149) at
org.sparkproject.guava.io.ByteStreams.read(ByteStreams.java:899) at
org.sparkproject.guava.io.ByteStreams.readFully(ByteStreams.java:733) at
org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.next(UnsafeRowSerializer.scala:127)
at
org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.next(UnsafeRowSerializer.scala:110)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:496) at
scala.collection.Iterator$$anon$10.next(Iterator.scala:461) at
org.apache.spark.util.CompletionIterator.next(CompletionIterator.scala:29) at
org.apache.spark.InterruptibleIterator.next(InterruptibleIterator.scala:40) at
scala.collection.Iterator$$anon$10.next(Iterator.scala:461) at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage7.sort_addToSorter_0$(Unknown
Source) at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage7.processNext(Unknown
Source) at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
at
org.apache.spark.sql.execution.RowIteratorFromScala.advanceNext(RowIterator.scala:82)
at
org.apache.spark.sql.execution.joins.SortMergeJoinScanner.advancedStreamed(SortMergeJoinExec.scala:1065)
at
org.apache.spark.sql.execution.joins.SortMergeJoinScanner.findNextOuterJoinRows(SortMergeJoinExec.scala:1024)
at
org.apache.spark.sql.execution.joins.OneSideOuterIterator.advanceStream(SortMergeJoinExec.scala:1201)
at
org.apache.spark.sql.execution.joins.OneSideOuterIterator.advanceNext(SortMergeJoinExec.scala:1240)
at
org.apache.spark.sql.execution.RowIteratorToScala.hasNext(RowIterator.scala:67)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.processNext(Unknown
Source) at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at
org.apache.spark.sql.execution.UnsafeExternalRowSorter.sort(UnsafeExternalRowSorter.java:225)
at
org.apache.spark.sql.execution.SortExec.$anonfun$doExecute$1(SortExec.scala:119)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:898) at
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:898)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:337) at
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) at
org.apache.spark.scheduler.Task.run(Task.scala:137) at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1510) at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at
[jira] [Updated] (SPARK-43242) shuffle diagnoseCorruption should not throw Unexpected type of BlockId for ShuffleBlockBatchId
[
https://issues.apache.org/jira/browse/SPARK-43242?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zhang Liang updated SPARK-43242:
Description:
Some of our spark app throw "Unexpected type of BlockId" exception as shown
below
According to BlockId.scala, we can found format such as
*{*}shuffle_12_5868_518_523{*}* is type of `ShuffleBlockBatchId`, which is not
handled properly in `ShuffleBlockFetcherIterator.diagnoseCorruption`.
Moreover, the new exception thrown in `diagnose` swallow the real exception in
certain cases. Since diagnoseCorruption is always used in exception handling as
a sidedish, I think it shouldn't throw exception at all
{code:java}
// code placeholder
23/03/07 03:01:24,485 [task-result-getter-1] WARN TaskSetManager: Lost task
104.0 in stage 36.0 (TID 6169): java.lang.IllegalArgumentException: Unexpected
type of BlockId, shuffle_12_5868_518_523 at
org.apache.spark.storage.ShuffleBlockFetcherIterator.diagnoseCorruption(ShuffleBlockFetcherIterator.scala:1079)at
org.apache.spark.storage.BufferReleasingInputStream.$anonfun$tryOrFetchFailedException$1(ShuffleBlockFetcherIterator.scala:1314)
at scala.Option.map(Option.scala:230)at
org.apache.spark.storage.BufferReleasingInputStream.tryOrFetchFailedException(ShuffleBlockFetcherIterator.scala:1313)
at
org.apache.spark.storage.BufferReleasingInputStream.read(ShuffleBlockFetcherIterator.scala:1299)
at java.io.BufferedInputStream.fill(BufferedInputStream.java:246) at
java.io.BufferedInputStream.read1(BufferedInputStream.java:286) at
java.io.BufferedInputStream.read(BufferedInputStream.java:345) at
java.io.DataInputStream.read(DataInputStream.java:149) at
org.sparkproject.guava.io.ByteStreams.read(ByteStreams.java:899) at
org.sparkproject.guava.io.ByteStreams.readFully(ByteStreams.java:733) at
org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.next(UnsafeRowSerializer.scala:127)
at
org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon$2$$anon$3.next(UnsafeRowSerializer.scala:110)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:496) at
scala.collection.Iterator$$anon$10.next(Iterator.scala:461) at
org.apache.spark.util.CompletionIterator.next(CompletionIterator.scala:29) at
org.apache.spark.InterruptibleIterator.next(InterruptibleIterator.scala:40) at
scala.collection.Iterator$$anon$10.next(Iterator.scala:461) at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage7.sort_addToSorter_0$(Unknown
Source) at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage7.processNext(Unknown
Source) at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
at
org.apache.spark.sql.execution.RowIteratorFromScala.advanceNext(RowIterator.scala:82)
at
org.apache.spark.sql.execution.joins.SortMergeJoinScanner.advancedStreamed(SortMergeJoinExec.scala:1065)
at
org.apache.spark.sql.execution.joins.SortMergeJoinScanner.findNextOuterJoinRows(SortMergeJoinExec.scala:1024)
at
org.apache.spark.sql.execution.joins.OneSideOuterIterator.advanceStream(SortMergeJoinExec.scala:1201)
at
org.apache.spark.sql.execution.joins.OneSideOuterIterator.advanceNext(SortMergeJoinExec.scala:1240)
at
org.apache.spark.sql.execution.RowIteratorToScala.hasNext(RowIterator.scala:67)
at
org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage9.processNext(Unknown
Source) at
org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at
org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) at
org.apache.spark.sql.execution.UnsafeExternalRowSorter.sort(UnsafeExternalRowSorter.java:225)
at
org.apache.spark.sql.execution.SortExec.$anonfun$doExecute$1(SortExec.scala:119)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:898) at
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:898)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) at
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373) at
org.apache.spark.rdd.RDD.iterator(RDD.scala:337) at
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) at
org.apache.spark.scheduler.Task.run(Task.scala:137) at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1510) at
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509) at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at
