incubator-uniffle icon indicating copy to clipboard operation
incubator-uniffle copied to clipboard
verified publisher icon apache

[Bug] Reading from hdfs file fails with writing hdfs file concurrent.

Open lwllvyb opened this issue 1 year ago • 3 comments

Code of Conduct

Search before asking

  • [X] I have searched in the issues and found no similar issues.

Describe the bug

When the shuffle stage finished, some blocks in server are being written to hdfs file, which will cause the client reading ops to fail.

The client reading ops log:

[21:08:05:679] [Executor task launch worker for task 881.1 in stage 53.0 (TID 104635)] INFO  org.apache.uniffle.client.impl.grpc.ShuffleServerGrpcNettyClient.getShuffleData:365 - GetShuffleData from x.x.x.x:17000 for appId[application_1716779728283_4017333_1718628854483], shuffleId[0], partitionId[881] cost 20 ms
[21:08:05:727] [Executor task launch worker for task 881.1 in stage 53.0 (TID 104635)] INFO  org.apache.uniffle.storage.handler.impl.HadoopClientReadHandler.init:156 - Find index file for shuffleId[0], partitionId[881] hdfs://xxx/rss/application_1716779728283_4017333_1718628854483/0/881-881/x.x.x.x-19977-17000_0_0.index
[21:08:05:746] [Executor task launch worker for task 881.1 in stage 53.0 (TID 104635)] WARN  org.apache.uniffle.storage.handler.impl.HadoopClientReadHandler.init:180 - Can't create ShuffleReaderHandler for hdfs://xxx/rss/application_1716779728283_4017333_1718628854483/0/881-881/x.x.x.x-19977-17000_0_0
org.apache.hadoop.hdfs.CannotObtainBlockLengthException: Cannot obtain block length for LocatedBlock{BP-1774150831-*-1647590905164:blk_26758443966_26343060354; getBlockSize()=4552; corrupt=false; offset=3623878656; locs=[DatanodeInfoWithStorage[*:9003,DS-af1dba43-5751-4bed-bea6-6050b3929bb4,DISK], DatanodeInfoWithStorage[*:9003,DS-d13fe05a-c894-4dc6-a2ec-2e43aae096c2,DISK]]} of /rss/application_1716779728283_4017333_1718628854483/0/881-881/*-19977-17000_0_0.data
	at org.apache.hadoop.hdfs.DFSInputStream.readBlockLength(DFSInputStream.java:455) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.hdfs.DFSInputStream.getLastBlockLength(DFSInputStream.java:364) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.hdfs.DFSInputStream.fetchLocatedBlocksAndGetLastBlockLength(DFSInputStream.java:345) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.hdfs.DFSInputStream.openInfo(DFSInputStream.java:242) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.hdfs.DFSInputStream.<init>(DFSInputStream.java:208) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.hdfs.DFSClient.openInternal(DFSClient.java:1159) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.hdfs.DFSClient.open(DFSClient.java:1122) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.hdfs.DistributedFileSystem$4.doCall(DistributedFileSystem.java:341) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.hdfs.DistributedFileSystem$4.doCall(DistributedFileSystem.java:337) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) ~[hadoop-common-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.hdfs.DistributedFileSystem.open(DistributedFileSystem.java:349) ~[hadoop-hdfs-client-3.2.1-tq-0.2.7.jar:?]
	at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:929) ~[hadoop-common-3.2.1-tq-0.2.7.jar:?]
	at org.apache.uniffle.storage.handler.impl.HadoopFileReader.createStream(HadoopFileReader.java:58) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.uniffle.storage.handler.impl.HadoopFileReader.<init>(HadoopFileReader.java:46) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.uniffle.storage.handler.impl.HadoopShuffleReadHandler.createHadoopReader(HadoopShuffleReadHandler.java:231) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.uniffle.storage.handler.impl.HadoopShuffleReadHandler.<init>(HadoopShuffleReadHandler.java:75) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.uniffle.storage.handler.impl.HadoopClientReadHandler.init(HadoopClientReadHandler.java:165) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.uniffle.storage.handler.impl.HadoopClientReadHandler.readShuffleData(HadoopClientReadHandler.java:199) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.uniffle.storage.handler.impl.ComposedClientReadHandler.readShuffleData(ComposedClientReadHandler.java:113) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.uniffle.storage.handler.impl.ComposedClientReadHandler.readShuffleData(ComposedClientReadHandler.java:134) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.uniffle.client.impl.ShuffleReadClientImpl.read(ShuffleReadClientImpl.java:308) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.uniffle.client.impl.ShuffleReadClientImpl.readShuffleBlockData(ShuffleReadClientImpl.java:216) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.spark.shuffle.reader.RssShuffleDataIterator.hasNext(RssShuffleDataIterator.java:115) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.shuffle.reader.RssShuffleReader$MultiPartitionIterator.hasNext(RssShuffleReader.java:312) ~[rss-client-spark3-shaded-0.9.0-SNAPSHOT.jar:?]
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at scala.collection.Iterator$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.15.jar:?]
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage17.sort_addToSorter_0$(Unknown Source) ~[?:?]
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage17.processNext(Unknown Source) ~[?:?]
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) ~[spark-sql_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.sql.execution.WholeStageCodegenExec$anon$1.hasNext(WholeStageCodegenExec.scala:760) ~[spark-sql_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.sql.execution.RowIteratorFromScala.advanceNext(RowIterator.scala:82) ~[spark-catalyst_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.sql.execution.joins.SortMergeJoinScanner.advancedBufferedToRowWithNullFreeJoinKey(SortMergeJoinExec.scala:1469) ~[spark-sql_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.sql.execution.joins.SortMergeJoinScanner.<init>(SortMergeJoinExec.scala:1329) ~[spark-sql_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.sql.execution.joins.SortMergeJoinExec.$anonfun$doExecute$1(SortMergeJoinExec.scala:229) ~[spark-sql_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.rdd.ZippedPartitionsRDD2.compute(ZippedPartitionsRDD.scala:89) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:329) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:329) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:329) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.scheduler.Task.run(Task.scala:136) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1556) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551) ~[spark-core_2.12-3.3.1.jar:3.3.1]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_392]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_392]
	at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_392]

The write op log in server

[2024-06-17 20:29:42.723] [DataStreamer for file /rss/application_1716779728283_4013066_1718626469818/0/846-846/*-19977-17000_0_0.data block BP-1774150831-*-1647590905164:blk_26757305915_26340993839] [WARN] DataStreamer.handleBadDatanode - Error Recovery for BP-1774150831-*-1647590905164:blk_26757305915_26340993839 in pipeline [DatanodeInfoWithStorage[*:9003,DS-18c43763-c3c5-49be-ae1e-69e5d762ff9e,DISK], DatanodeInfoWithStorage[*:9003,DS-52eddd2a-84d4-449b-832f-4c670349dac1,DISK]]: datanode 0(DatanodeInfoWithStorage[*:9003,DS-18c43763-c3c5-49be-ae1e-69e5d762ff9e,DISK]) is bad.
[2024-06-17 20:29:52.259] [Grpc-1212] [INFO] ShuffleServerGrpcService.appHeartbeat - Get heartbeat from application_1716779728283_4013066_1718626469818
[2024-06-17 20:29:59.384] [DataStreamer for file /rss/application_1716779728283_4013066_1718626469818/0/840-840/*-19977-17000_0_0.data block BP-1774150831-*-1647590905164:blk_26757307918_26340996470] [INFO] DataStreamer.createBlockOutputStream - Exception in createBlockOutputStream blk_26757307918_26340996470
java.net.SocketTimeoutException: 20000 millis timeout while waiting for channel to be ready for read. ch : java.nio.channels.SocketChannel[connected local=/*:50760 remote=/*:9003]
        at org.apache.hadoop.net.SocketIOWithTimeout.doIO(SocketIOWithTimeout.java:164)
        at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:161)
        at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:131)
        at org.apache.hadoop.net.SocketInputStream.read(SocketInputStream.java:118)
        at java.io.FilterInputStream.read(FilterInputStream.java:83)
        at java.io.FilterInputStream.read(FilterInputStream.java:83)
        at org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed(PBHelperClient.java:459)
        at org.apache.hadoop.hdfs.DataStreamer.createBlockOutputStream(DataStreamer.java:1776)
        at org.apache.hadoop.hdfs.DataStreamer.setupPipelineInternal(DataStreamer.java:1513)
        at org.apache.hadoop.hdfs.DataStreamer.setupPipelineForAppendOrRecovery(DataStreamer.java:1487)
        at org.apache.hadoop.hdfs.DataStreamer.processDatanodeOrExternalError(DataStreamer.java:1262)
        at org.apache.hadoop.hdfs.DataStreamer.run(DataStreamer.java:673)
[2024-06-17 20:30:00.824] [HadoopFlushEventThreadPool-9] [WARN] SingleStorageManager.write - Exception happened when write data for ShuffleDataFlushEvent: eventId=8342, appId=application_1716779728283_4013066_1718626469818, shuffleId=0, startPartition=840, endPartition=840, retryTimes=0, underStorage=HadoopStorage, isPended=false, ownedByHugePartition=false, try again
java.lang.IllegalArgumentException: Self-suppression not permitted
        at java.lang.Throwable.addSuppressed(Throwable.java:1072)
        at org.apache.uniffle.storage.handler.impl.HadoopShuffleWriteHandler.write(HadoopShuffleWriteHandler.java:147)
        at org.apache.uniffle.storage.handler.impl.PooledHadoopShuffleWriteHandler.write(PooledHadoopShuffleWriteHandler.java:122)
        at org.apache.uniffle.server.storage.SingleStorageManager.write(SingleStorageManager.java:59)
        at org.apache.uniffle.server.storage.HybridStorageManager.write(HybridStorageManager.java:130)
        at org.apache.uniffle.server.ShuffleFlushManager.processFlushEvent(ShuffleFlushManager.java:165)
        at org.apache.uniffle.server.DefaultFlushEventHandler.handleEventAndUpdateMetrics(DefaultFlushEventHandler.java:97)
        at org.apache.uniffle.server.DefaultFlushEventHandler.lambda$dispatchEvent$0(DefaultFlushEventHandler.java:219)
        at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1640)
        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)
Caused by: java.io.IOException: All datanodes [DatanodeInfoWithStorage[*:9003,DS-52eddd2a-84d4-449b-832f-4c670349dac1,DISK]] are bad. Aborting...
        at org.apache.hadoop.hdfs.DataStreamer.handleBadDatanode(DataStreamer.java:1567)
        at org.apache.hadoop.hdfs.DataStreamer.setupPipelineInternal(DataStreamer.java:1501)
        at org.apache.hadoop.hdfs.DataStreamer.setupPipelineForAppendOrRecovery(DataStreamer.java:1487)
        at org.apache.hadoop.hdfs.DataStreamer.processDatanodeOrExternalError(DataStreamer.java:1262)
        at org.apache.hadoop.hdfs.DataStreamer.run(DataStreamer.java:673)
[2024-06-17 20:30:01.831] [FallBackFlushEventThreadPool-0] [WARN] HadoopShuffleWriteHandler.write - Write failed with 2786 blocks for *-19977-17000_0_0
org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException): Failed to APPEND_FILE /rss/application_1716779728283_4013066_1718626469818/0/840-840/*-19977-17000_0_0.data for DFSClient_NONMAPREDUCE_-1504047525_40 on * because DFSClient_NONMAPREDUCE_-1504047525_40 is already the current lease holder.
        at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.recoverLeaseInternal(FSNamesystem.java:2633)
        at org.apache.hadoop.hdfs.server.namenode.FSDirAppendOp.appendFile(FSDirAppendOp.java:122)
        at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.appendFile(FSNamesystem.java:2717)
        at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.append(NameNodeRpcServer.java:880)
        at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.append(ClientNamenodeProtocolServerSideTranslatorPB.java:351)
        at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
        at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:529)
        at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:1073)
        at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:1039)
        at org.apache.hadoop.ipc.Server$RpcCall.run(Server.java:963)
        at java.base/java.security.AccessController.doPrivileged(Native Method)
        at java.base/javax.security.auth.Subject.doAs(Subject.java:423)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:2065)
        at org.apache.hadoop.ipc.Server$Handler.run(Server.java:3047)

Affects Version(s)

master

Uniffle Server Log Output

No response

Uniffle Engine Log Output

No response

Uniffle Server Configurations

No response

Uniffle Engine Configurations

No response

Additional context

No response

Are you willing to submit PR?

  • [X] Yes I am willing to submit a PR!

lwllvyb avatar Oct 09 '24 02:10 lwllvyb

What's your solution?

jerqi avatar Oct 09 '24 03:10 jerqi

Pleae attach your doc here @lwllvyb

zuston avatar Oct 10 '24 01:10 zuston

https://docs.google.com/document/d/1Lb2D1FsZISXSKJO1WjyRJN2ymu_xiOFcVmKmYkoNmpQ/edit#heading=h.deydo9nszbc I can add more details if necessary. @jerqi

lwllvyb avatar Oct 10 '24 03:10 lwllvyb