lwllvyb opened a new issue, #2162: URL: https://github.com/apache/incubator-uniffle/issues/2162
### Code of Conduct - [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) ### Search before asking - [X] I have searched in the [issues](https://github.com/apache/incubator-uniffle/issues?q=is%3Aissue) 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! -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
