search

apache / incubator-uniffle

Uniffle is a high performance, general purpose Remote Shuffle Service.
https://uniffle.apache.org/
Apache License 2.0
385 stars 149 forks source link

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

Open lwllvyb opened 1 month ago

lwllvyb commented 1 month ago

Code of Conduct

Search before asking

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?

jerqi commented 1 month ago

What's your solution?

zuston commented 1 month ago

Pleae attach your doc here @lwllvyb

lwllvyb commented 1 month ago

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