Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

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

Open
3 tasks done
lwllvyb opened this issue Oct 9, 2024 · 3 comments
Open
3 tasks done

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

lwllvyb opened this issue Oct 9, 2024 · 3 comments

Comments

@lwllvyb
Copy link
Contributor

lwllvyb commented Oct 9, 2024

Code of Conduct

Search before asking

  • 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?

  • Yes I am willing to submit a PR!
@jerqi
Copy link
Contributor

jerqi commented Oct 9, 2024

What's your solution?

@zuston
Copy link
Member

zuston commented Oct 10, 2024

Pleae attach your doc here @lwllvyb

@lwllvyb
Copy link
Contributor Author

lwllvyb commented Oct 10, 2024

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

No branches or pull requests
3 participants
@jerqi @zuston @lwllvyb