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

### 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.(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.(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.(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.(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!

Start by tracing HadoopClientReadHandler.init and HadoopFileReader.createStream on the read path, then inspect HadoopShuffleWriteHandler.write and PooledHadoopShuffleWriteHandler.write on the concurrent write path. Reproduce the HDFS read failure during an active write and determine the synchronization or visibility boundary needed; done means reads no longer fail while shuffle data is being written, with regression coverage for the reported case.

Written by the indexing model from the issue text.