You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "DapengShi (via GitHub)" <gi...@apache.org> on 2024/08/22 09:52:34 UTC
[I] deadlock when spark call delete row postition [iceberg]
DapengShi opened a new issue, #10987:
URL: https://github.com/apache/iceberg/issues/10987
### Apache Iceberg version
main (development)
### Query engine
Spark
### Please describe the bug 🐞
we ran spark on k8s and this dead lock happened when we set spark exectuor cpu to 1
here is the key thread dump of the executor
```
"Executor task launch worker for task 39128.1 in stage 1.0 (TID 52199)" #5788 daemon prio=5 os_prio=0 tid=0x00007f1964042800 nid=0x16d3 sleeping[0x00007f190adfd000]
java.lang.Thread.State: TIMED_WAITING (sleeping)
at java.lang.Thread.sleep(Native Method)
at org.apache.iceberg.util.Tasks.waitFor(Tasks.java:518)
at org.apache.iceberg.util.Tasks.access$800(Tasks.java:42)
at org.apache.iceberg.util.Tasks$Builder.runParallel(Tasks.java:358)
at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:201)
at org.apache.iceberg.util.Tasks$Builder.run(Tasks.java:196)
at org.apache.iceberg.data.BaseDeleteLoader.execute(BaseDeleteLoader.java:236)
at org.apache.iceberg.data.BaseDeleteLoader.loadPositionDeletes(BaseDeleteLoader.java:150)
at org.apache.iceberg.data.DeleteFilter.deletedRowPositions(DeleteFilter.java:227)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader$ColumnBatchLoader.posDelRowIdMapping(ColumnarBatchReader.java:156)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader$ColumnBatchLoader.initRowIdMapping(ColumnarBatchReader.java:144)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader$ColumnBatchLoader.loadDataToColumnBatch(ColumnarBatchReader.java:96)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader.read(ColumnarBatchReader.java:72)
at org.apache.iceberg.spark.data.vectorized.ColumnarBatchReader.read(ColumnarBatchReader.java:44)
at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.next(VectorizedParquetReader.java:147)
at org.apache.iceberg.spark.source.BaseReader.next(BaseReader.java:138)
at org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:120)
at org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:158)
at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63)
at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63)
at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1$$Lambda$862/982683868.apply(Unknown Source)
at scala.Option.exists(Option.scala:376)
at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)
at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.advanceToNextIter(DataSourceRDD.scala:97)
at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.columnartorow_nextBatch_0$(Unknown Source)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:43)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:140)
at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:104)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:54)
at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:161)
at org.apache.spark.scheduler.Task.run(Task.scala:141)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620)
at org.apache.spark.executor.Executor$TaskRunner$$Lambda$508/1366295625.apply(Unknown Source)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623)
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)
Locked ownable synchronizers:
- <0x00000004550025b8> (a java.util.concurrent.ThreadPoolExecutor$Worker)
...
"ForkJoinPool.commonPool-worker-0" #5789 daemon prio=5 os_prio=0 tid=0x00007f18ac005800 nid=0x16d4 waiting for monitor entry [0x00007f190b9fe000]
java.lang.Thread.State: BLOCKED (on object monitor)
at java.util.concurrent.ConcurrentHashMap.computeIfPresent(ConcurrentHashMap.java:1760)
- waiting to lock <0x0000000546031ec8> (a java.util.concurrent.ConcurrentHashMap$Node)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache.evictEntry(BoundedLocalCache.java:912)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache.expireAfterAccessEntries(BoundedLocalCache.java:820)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache.expireAfterAccessEntries(BoundedLocalCache.java:806)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache.expireEntries(BoundedLocalCache.java:784)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache.maintenance(BoundedLocalCache.java:1491)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache.performCleanUp(BoundedLocalCache.java:1460)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache$PerformCleanupTask.run(BoundedLocalCache.java:3359)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache$PerformCleanupTask.exec(BoundedLocalCache.java:3346)
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:175)
Locked ownable synchronizers:
- <0x0000000481f0eb08> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)
...
"iceberg-delete-worker-pool-3" #85 daemon prio=5 os_prio=0 tid=0x00007f1974fd1800 nid=0x8e in Object.wait() [0x00007f1949dfd000]
java.lang.Thread.State: WAITING (on object monitor)
at java.lang.Object.wait(Native Method)
at java.util.concurrent.ForkJoinTask.externalAwaitDone(ForkJoinTask.java:334)
- locked <0x0000000455002648> (a java.util.stream.ReduceOps$ReduceTask)
at java.util.concurrent.ForkJoinTask.doInvoke(ForkJoinTask.java:405)
at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:734)
at java.util.stream.ReduceOps$ReduceOp.evaluateParallel(ReduceOps.java:714)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:566)
at org.apache.hadoop.fs.statistics.impl.EvaluatingStatisticsMap.entrySet(EvaluatingStatisticsMap.java:166)
- locked <0x00000004550026a0> (a org.apache.hadoop.fs.statistics.impl.EvaluatingStatisticsMap)
at java.util.Collections$UnmodifiableMap.entrySet(Collections.java:1483)
at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.copyMap(IOStatisticsBinding.java:173)
at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.snapshotMap(IOStatisticsBinding.java:219)
at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.snapshotMap(IOStatisticsBinding.java:202)
at org.apache.hadoop.fs.statistics.IOStatisticsSnapshot.snapshot(IOStatisticsSnapshot.java:163)
- locked <0x0000000455002728> (a org.apache.hadoop.fs.statistics.IOStatisticsSnapshot)
at org.apache.hadoop.fs.statistics.IOStatisticsSnapshot.<init>(IOStatisticsSnapshot.java:125)
at org.apache.hadoop.fs.statistics.IOStatisticsSupport.snapshotIOStatistics(IOStatisticsSupport.java:49)
at org.apache.hadoop.fs.s3a.S3AInstrumentation$InputStreamStatistics.merge(S3AInstrumentation.java:1220)
at org.apache.hadoop.fs.s3a.S3AInstrumentation$InputStreamStatistics.close(S3AInstrumentation.java:1126)
at org.apache.hadoop.fs.s3a.S3AInputStream.close(S3AInputStream.java:620)
- locked <0x0000000455002770> (a org.apache.hadoop.fs.s3a.S3AInputStream)
at java.io.FilterInputStream.close(FilterInputStream.java:181)
at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.close(DelegatingSeekableInputStream.java:50)
at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader.close(ParquetFileReader.java:1447)
at org.apache.iceberg.parquet.ParquetReader$FileIterator.close(ParquetReader.java:161)
at org.apache.iceberg.io.CloseableGroup.close(CloseableGroup.java:80)
at org.apache.iceberg.deletes.Deletes.toPositionIndexes(Deletes.java:154)
at org.apache.iceberg.data.BaseDeleteLoader.readPosDeletes(BaseDeleteLoader.java:169)
at org.apache.iceberg.data.BaseDeleteLoader.lambda$getOrReadPosDeletes$3(BaseDeleteLoader.java:160)
at org.apache.iceberg.data.BaseDeleteLoader$$Lambda$1035/194701756.get(Unknown Source)
at org.apache.iceberg.spark.SparkExecutorCache.lambda$loadFunc$0(SparkExecutorCache.java:122)
at org.apache.iceberg.spark.SparkExecutorCache$$Lambda$1042/979932856.apply(Unknown Source)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.LocalCache.lambda$statsAware$0(LocalCache.java:139)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.LocalCache$$Lambda$1046/1506524809.apply(Unknown Source)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache.lambda$doComputeIfAbsent$14(BoundedLocalCache.java:2431)
- locked <0x000000054a1eb350> (a org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.PSAMW)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache$$Lambda$867/1581302069.apply(Unknown Source)
at java.util.concurrent.ConcurrentHashMap.compute(ConcurrentHashMap.java:1877)
- locked <0x0000000546031ec8> (a java.util.concurrent.ConcurrentHashMap$Node)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache.doComputeIfAbsent(BoundedLocalCache.java:2404)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.BoundedLocalCache.computeIfAbsent(BoundedLocalCache.java:2387)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.LocalCache.computeIfAbsent(LocalCache.java:108)
at org.apache.iceberg.shaded.com.github.benmanes.caffeine.cache.LocalManualCache.get(LocalManualCache.java:62)
at org.apache.iceberg.spark.SparkExecutorCache.getOrLoad(SparkExecutorCache.java:114)
at org.apache.iceberg.spark.source.BaseReader$SparkDeleteFilter$CachingDeleteLoader.getOrLoad(BaseReader.java:297)
at org.apache.iceberg.data.BaseDeleteLoader.getOrReadPosDeletes(BaseDeleteLoader.java:160)
at org.apache.iceberg.data.BaseDeleteLoader.lambda$loadPositionDeletes$2(BaseDeleteLoader.java:150)
at org.apache.iceberg.data.BaseDeleteLoader$$Lambda$1032/2004566363.apply(Unknown Source)
at org.apache.iceberg.data.BaseDeleteLoader.lambda$execute$7(BaseDeleteLoader.java:236)
at org.apache.iceberg.data.BaseDeleteLoader$$Lambda$1034/1084935997.run(Unknown Source)
at org.apache.iceberg.util.Tasks$Builder.runTaskWithRetry(Tasks.java:413)
at org.apache.iceberg.util.Tasks$Builder.access$300(Tasks.java:69)
at org.apache.iceberg.util.Tasks$Builder$1.run(Tasks.java:315)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
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)
Locked ownable synchronizers:
- <0x0000000481e9ec38> (a java.util.concurrent.ThreadPoolExecutor$Worker)
"iceberg-delete-worker-pool-2" #84 daemon prio=5 os_prio=0 tid=0x00007f1974fcf800 nid=0x8d waiting on condition [0x00007f1949efe000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x0000000481e84b48> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Locked ownable synchronizers:
- None
"iceberg-delete-worker-pool-1" #83 daemon prio=5 os_prio=0 tid=0x00007f1974fce000 nid=0x8c waiting on condition [0x00007f1949ffe000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x0000000481e84b48> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Locked ownable synchronizers:
- None
"iceberg-delete-worker-pool-0" #82 daemon prio=5 os_prio=0 tid=0x00007f1974fcc000 nid=0x8b waiting on condition [0x00007f194a7fe000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x0000000481e84b48> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
Locked ownable synchronizers:
- None
```
As we can see, the task thread try to call `org.apache.iceberg.data.DeleteFilter.deletedRowPositions` and it waited all iceberg-delete-worker-pool threads to finish tasks. And there was only one task left in `iceberg-delete-worker-pool-3`, and the task called `BoundedLocalCache.computeIfAbsent`, as `ForkJoinPool.commonPool-worker-0` implied maybe BoundedLocalCache trigged one clean up task during this `computeIfAbsent`. And `BoundedLocalCache` uses ForkJoin common pool as default executor to execute this clean up task. The task was blocked by `ConcurrentHashMap`. So it waited `iceberg-delete-worker-pool-3` to finish. However in `BoundedLocalCache.computeIfAbsent` of `iceberg-delete-worker-pool-3`, the `org.apache.hadoop.fs.statistics.impl.EvaluatingStatisticsMap.entrySet` was called, and it uses `parallelStream()` to transform the set. `parallelStream()` will also calls ForkJoin common pool to execute task. Because the spark executor cpu was set to 1, so there would be only on thr
ead in ForkJoin common pool, which was blocked by `BoundedLocalCache` clean up task and could not finish the `parallelStream()` task. Dead lock happend.
### Willingness to contribute
- [ ] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time
--
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: issues-unsubscribe@iceberg.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] deadlock when spark call delete row postition [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed issue #10987: deadlock when spark call delete row postition
URL: https://github.com/apache/iceberg/issues/10987
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] deadlock when spark call delete row postition [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #10987:
URL: https://github.com/apache/iceberg/issues/10987#issuecomment-2699325005
This issue has been closed because it has not received any activity in the last 14 days since being marked as 'stale'
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] deadlock when spark call delete row postition [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #10987:
URL: https://github.com/apache/iceberg/issues/10987#issuecomment-2667209074
This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when 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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org