[CELEBORN-980] Asynchronously delete original files to fix `ReusedExchange` bug by cfmcgrady · Pull Request #1932 · apache/celeborn

cfmcgrady · 2023-09-21T07:52:30Z

What changes were proposed in this pull request?

The ReusedExchange operator has the potential to generate different types of fetch requests, including both non-range and range requests. Currently, an issue arises due to the synchronous deletion of the original file by the Celeborn worker upon completion of sorting. This issue leads to the failure of non-range requests following a range request for the same partition.

the snippets to reproduce this bug

  val sparkConf = new SparkConf().setAppName("celeborn-test").setMaster("local[2]")
    .set("spark.shuffle.manager", "org.apache.spark.shuffle.celeborn.SparkShuffleManager")
    .set(s"spark.${CelebornConf.MASTER_ENDPOINTS.key}", masterInfo._1.rpcEnv.address.toString)
    .set("spark.sql.autoBroadcastJoinThreshold", "-1")
    .set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "100")
    .set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "100")
  val spark = SparkSession.builder()
    .config(sparkConf)
    .getOrCreate()
  spark.range(0, 1000, 1, 10)
    .selectExpr("id as k1", "id as v1")
    .createOrReplaceTempView("ta")
  spark.range(0, 1000, 1, 10)
    .selectExpr("id % 1 as k21", "id % 1 as k22", "id as v2")
    .createOrReplaceTempView("tb")
  spark.range(140)
    .select(
      col("id").cast("long").as("k3"),
      concat(col("id").cast("string"), lit("a")).as("v3"))
    .createOrReplaceTempView("tc")

  spark.sql(
    """
      |SELECT *
      |FROM ta
      |LEFT JOIN tb ON ta.k1 = tb.k21
      |LEFT JOIN tc ON tb.k22 = tc.k3
      |""".stripMargin)
    .createOrReplaceTempView("v1")

  spark.sql(
    """
      |SELECT * FROM v1 WHERE v3 IS NOT NULL
      |UNION
      |SELECT * FROM v1
      |""".stripMargin)
    .collect()

This PR proposes a solution to address this problem. It introduces an asynchronous thread for the removal of the original file. Once the sorted file is generated for a given partition, this modification ensures that both non-range and range fetch requests will be able to and only fetch the sorted file once it is generated for a given partition.

this activity diagram of openStream

Does this PR introduce any user-facing change?

No, only bug fix

How was this patch tested?

UT

style

…g-v4

codecov · 2023-09-21T08:04:40Z

Codecov Report

Merging #1932 (6c86e01) into main (2407cae) will increase coverage by 0.41%.
Report is 22 commits behind head on main.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##             main    #1932      +/-   ##
==========================================
+ Coverage   46.57%   46.98%   +0.41%     
==========================================
  Files         164      164              
  Lines       10293    10361      +68     
  Branches      938      956      +18     
==========================================
+ Hits         4793     4867      +74     
+ Misses       5185     5177       -8     
- Partials      315      317       +2

Files	Coverage Δ
...java/org/apache/celeborn/common/meta/FileInfo.java	`53.00% <100.00%> (+9.63%)`	⬆️
...eleborn/common/network/client/TransportClient.java	`39.86% <ø> (ø)`
...eborn/common/network/protocol/BufferStreamEnd.java	`0.00% <ø> (ø)`
...born/common/network/protocol/TransportMessage.java	`68.58% <ø> (+8.58%)`	⬆️
...he/celeborn/common/util/ShuffleBlockInfoUtils.java	`86.85% <100.00%> (+1.13%)`	⬆️

... and 13 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

cfmcgrady · 2023-09-21T08:25:21Z

As suggested #1907 (review), implemented a new approach that moves the state (stream ids) responsible for deleting unsorted files into the FileInfo class. cc @waitinfuture @FMX

cfmcgrady · 2023-09-21T08:28:29Z

the openStream has become complex enough.

FMX · 2023-09-22T03:01:13Z

...er/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java

+                  while (it.hasNext()) {
+                    PartitionFilesSorter.FileSorter sorter = it.next();
+                    try {
+                      if (sorter.getOriginFileInfo().isStreamsEmpty()) {


If there is no non skew streams, the origin file can be deleted. I think it's good to record all stream in fileinfo, I'll need it in supporting memory storage.

If there is no non skew streams, the origin file can be deleted.

it's not, only when the range open stream request(skew stream) will trigger FileSorter.sort(), and then the FileSorter instance will be added to the cleanup queue.

I think it's good to record all stream in fileinfo, I'll need it in supporting memory storage.

yeah, I also read the proposal you posted on the mailing list, in the current implementation, record all activity streams in the fileinfo already done.

the relations between stream and fileinfo:

for non-skew streams(non-range open stream request), it's n to 1:

non-range stream id1 non-range stream id2 ---> original unsorted FileInfo A non-range stream id3

for skew streams(range open stream request), it's 1 to 1, this is because we don't manage the sorted fileInfos within StorageManager:

range stream id4 ---> sorted FileInfo B range stream id5 ---> sorted FileInfo C range stream id6 ---> sorted FileInfo D

worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala

FMX · 2023-09-22T03:47:53Z

common/src/main/java/org/apache/celeborn/common/meta/FileInfo.java

  }
+
+  public void setSorted() {
+    synchronized (sorted) {


This syncrhonize block is redundant.

it's for blocking the addStream

I agree with @FMX , the synchronization in addStream and isStreamsEmpty are necessary because these two methods need to be sequential, but it's unnecessary to synchronize in setSorted and closeStream.

FMX · 2023-09-22T03:48:20Z

common/src/main/java/org/apache/celeborn/common/meta/FileInfo.java

+  }
+
+  public void closeStream(long streamId) {
+    synchronized (sorted) {


…g-v4

waitinfuture · 2023-10-07T08:14:33Z

...er/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java

    UserIdentifier userIdentifier = fileInfo.getUserIdentifier();
-
+    // not a range read and the sorted file not generate now
+    if (endMapIndex == Integer.MAX_VALUE && !sortedShuffleFiles.containsKey(shuffleKey)) {


To check whether the sorted file has been generated, should also check the sorted set

if (endMapIndex == Integer.MAX_VALUE && !sortedShuffleFiles.containsKey(shuffleKey) && !sortedShuffleFiles.get(shuffleKey).contains(fileId)) { return fileInfo; }

Or else, do not call getSortedFileInfo if endMapIndex is max int.

improved using Option 2.

waitinfuture · 2023-10-07T08:17:47Z

worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala

+            startIndex,
+            endIndex)
+          // non-range openStream request
+          if (endIndex == Int.MaxValue && !fileInfo.addStream(streamId)) {


getSortedFileInfo may return the sorted file info even though endIndex == Int.MaxValue. IMO here we can just call original fileInfo's addStream like following:

var fileInfo = getRawFileInfo(shuffleKey, fileName) fileInfo.getPartitionType match { case PartitionType.REDUCE => val streamId = chunkStreamManager.nextStreamId() // non-range openStream request if (endIndex == Int.MaxValue && !fileInfo.addStream(streamId)) { // This branch is entered when the original unsorted file has been sorted by another // range's openStream request. retry fetching the sorted FileInfo. fileInfo = partitionsSorter.getSortedFileInfo( shuffleKey, fileName, fileInfo, startIndex, endIndex) assert(Utils.isSortedPath(fileInfo.getFilePath)) }

we also need to return the sorted file information for range openStream requests.

addressed in 6c86e01

cfmcgrady · 2023-10-09T02:31:57Z

encountered another problem while running the ReusedExchangeSuite after making this change. Although the suite passed.

seems the worker can't exit correctly.

23/10/09 09:57:26,422 ERROR [fetch-server-52-2] ResourceLeakDetector: LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
Created at:
	io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:403)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
	io.netty.buffer.AbstractByteBufAllocator.ioBuffer(AbstractByteBufAllocator.java:140)
	io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator$MaxMessageHandle.allocate(DefaultMaxMessagesRecvByteBufAllocator.java:120)
	io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:150)
	io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
	io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
	io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
	io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:750)
23/10/09 09:57:26,422 ERROR [data-client-79-4] ResourceLeakDetector: LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
Created at:
	io.netty.buffer.AbstractByteBufAllocator.compositeDirectBuffer(AbstractByteBufAllocator.java:224)
	io.netty.buffer.AbstractByteBufAllocator.compositeBuffer(AbstractByteBufAllocator.java:202)
	org.apache.celeborn.common.network.util.TransportFrameDecoder.decodeNext(TransportFrameDecoder.java:143)
	org.apache.celeborn.common.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:66)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
	io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
	io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
	io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:750)


23/10/09 09:57:32,109 ERROR [shutdown-hook-0] MasterClient: Send rpc with failure, has tried 0, max try 15!
org.apache.celeborn.common.exception.CelebornException: Exception thrown in awaitResult: 
	at org.apache.celeborn.common.util.ThreadUtils$.awaitResult(ThreadUtils.scala:229)
	at org.apache.celeborn.common.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:74)
	at org.apache.celeborn.common.client.MasterClient.sendMessageInner(MasterClient.java:151)
	at org.apache.celeborn.common.client.MasterClient.askSync(MasterClient.java:123)
	at org.apache.celeborn.service.deploy.worker.Worker.exitImmediately(Worker.scala:691)
	at org.apache.celeborn.service.deploy.worker.Worker$$anon$5.run(Worker.scala:714)
	at java.lang.Thread.run(Thread.java:750)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	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.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@6cbccb6c rejected from java.util.concurrent.ScheduledThreadPoolExecutor@13bac7f3[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
	at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:326)
	at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:533)
	at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.ask(NettyRpcEnv.scala:244)
	at org.apache.celeborn.common.rpc.netty.NettyRpcEndpointRef.ask(NettyRpcEnv.scala:418)
	at org.apache.celeborn.common.client.MasterClient.sendMessageInner(MasterClient.java:150)
	... 9 more
23/10/09 09:57:32,110 ERROR [shutdown-hook-0] Worker: Fail report to master, need wait PartitionLocation auto release: 

 Partition Location Info:
 primary: Map(local-1696816632431-1 -> {}, local-1696816632431-2 -> {}, local-1696816632431-0 -> {}, local-1696816632431-5 -> {}, local-1696816632431-3 -> {}, local-1696816632431-4 -> {})
 replica: Map()

org.apache.celeborn.common.exception.CelebornException: Exception thrown in awaitResult: 
	at org.apache.celeborn.common.util.ThreadUtils$.awaitResult(ThreadUtils.scala:229)
	at org.apache.celeborn.common.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:74)
	at org.apache.celeborn.common.client.MasterClient.sendMessageInner(MasterClient.java:151)
	at org.apache.celeborn.common.client.MasterClient.askSync(MasterClient.java:123)
	at org.apache.celeborn.service.deploy.worker.Worker.exitImmediately(Worker.scala:691)
	at org.apache.celeborn.service.deploy.worker.Worker$$anon$5.run(Worker.scala:714)
	at java.lang.Thread.run(Thread.java:750)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	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.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@6cbccb6c rejected from java.util.concurrent.ScheduledThreadPoolExecutor@13bac7f3[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
	at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:326)
	at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:533)
	at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.ask(NettyRpcEnv.scala:244)
	at org.apache.celeborn.common.rpc.netty.NettyRpcEndpointRef.ask(NettyRpcEnv.scala:418)
	at org.apache.celeborn.common.client.MasterClient.sendMessageInner(MasterClient.java:150)
	... 9 more

I'll investigate further later on.

waitinfuture

LGTM, thanks! Merged to main(v0.4.0)

FMX · 2023-10-09T04:33:57Z

encountered another problem while running the ReusedExchangeSuite after making this change. Although the suite passed.

seems the worker can't exit correctly.

23/10/09 09:57:26,422 ERROR [fetch-server-52-2] ResourceLeakDetector: LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
Created at:
	io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:403)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:188)
	io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:179)
	io.netty.buffer.AbstractByteBufAllocator.ioBuffer(AbstractByteBufAllocator.java:140)
	io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator$MaxMessageHandle.allocate(DefaultMaxMessagesRecvByteBufAllocator.java:120)
	io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:150)
	io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
	io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
	io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
	io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:750)
23/10/09 09:57:26,422 ERROR [data-client-79-4] ResourceLeakDetector: LEAK: ByteBuf.release() was not called before it's garbage-collected. See https://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records: 
Created at:
	io.netty.buffer.AbstractByteBufAllocator.compositeDirectBuffer(AbstractByteBufAllocator.java:224)
	io.netty.buffer.AbstractByteBufAllocator.compositeBuffer(AbstractByteBufAllocator.java:202)
	org.apache.celeborn.common.network.util.TransportFrameDecoder.decodeNext(TransportFrameDecoder.java:143)
	org.apache.celeborn.common.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:66)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
	io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
	io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
	io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
	io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
	io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
	io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
	io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
	io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	java.lang.Thread.run(Thread.java:750)


23/10/09 09:57:32,109 ERROR [shutdown-hook-0] MasterClient: Send rpc with failure, has tried 0, max try 15!
org.apache.celeborn.common.exception.CelebornException: Exception thrown in awaitResult: 
	at org.apache.celeborn.common.util.ThreadUtils$.awaitResult(ThreadUtils.scala:229)
	at org.apache.celeborn.common.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:74)
	at org.apache.celeborn.common.client.MasterClient.sendMessageInner(MasterClient.java:151)
	at org.apache.celeborn.common.client.MasterClient.askSync(MasterClient.java:123)
	at org.apache.celeborn.service.deploy.worker.Worker.exitImmediately(Worker.scala:691)
	at org.apache.celeborn.service.deploy.worker.Worker$$anon$5.run(Worker.scala:714)
	at java.lang.Thread.run(Thread.java:750)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	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.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@6cbccb6c rejected from java.util.concurrent.ScheduledThreadPoolExecutor@13bac7f3[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
	at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:326)
	at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:533)
	at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.ask(NettyRpcEnv.scala:244)
	at org.apache.celeborn.common.rpc.netty.NettyRpcEndpointRef.ask(NettyRpcEnv.scala:418)
	at org.apache.celeborn.common.client.MasterClient.sendMessageInner(MasterClient.java:150)
	... 9 more
23/10/09 09:57:32,110 ERROR [shutdown-hook-0] Worker: Fail report to master, need wait PartitionLocation auto release: 

 Partition Location Info:
 primary: Map(local-1696816632431-1 -> {}, local-1696816632431-2 -> {}, local-1696816632431-0 -> {}, local-1696816632431-5 -> {}, local-1696816632431-3 -> {}, local-1696816632431-4 -> {})
 replica: Map()

org.apache.celeborn.common.exception.CelebornException: Exception thrown in awaitResult: 
	at org.apache.celeborn.common.util.ThreadUtils$.awaitResult(ThreadUtils.scala:229)
	at org.apache.celeborn.common.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:74)
	at org.apache.celeborn.common.client.MasterClient.sendMessageInner(MasterClient.java:151)
	at org.apache.celeborn.common.client.MasterClient.askSync(MasterClient.java:123)
	at org.apache.celeborn.service.deploy.worker.Worker.exitImmediately(Worker.scala:691)
	at org.apache.celeborn.service.deploy.worker.Worker$$anon$5.run(Worker.scala:714)
	at java.lang.Thread.run(Thread.java:750)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	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.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@6cbccb6c rejected from java.util.concurrent.ScheduledThreadPoolExecutor@13bac7f3[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
	at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:326)
	at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:533)
	at org.apache.celeborn.common.rpc.netty.NettyRpcEnv.ask(NettyRpcEnv.scala:244)
	at org.apache.celeborn.common.rpc.netty.NettyRpcEndpointRef.ask(NettyRpcEnv.scala:418)
	at org.apache.celeborn.common.client.MasterClient.sendMessageInner(MasterClient.java:150)
	... 9 more

I'll investigate further later on.

Hi, I've seen this several times. IMO, this exception occurs when a fetch handler decodes some RPC messages while the server is ending. As you can see this exception occurs close to the shutdown hook's log.

…oveOriginalFiles.enabled` ### What changes were proposed in this pull request? As title ### Why are the changes needed? The config key `celeborn.worker.sortPartition.eagerlyRemoveOriginalFiles.enabled` has become unnecessary as a result of #1932 ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Pass GA Closes #1999 from cfmcgrady/celeborn-1047. Authored-by: Fu Chen <cfmcgrady@gmail.com> Signed-off-by: zky.zhoukeyong <zky.zhoukeyong@alibaba-inc.com>

…ger to close stream for ReusedExchange ### What changes were proposed in this pull request? `OpenStream` should register stream via `ChunkStreamManager`, which is served to obtain disk file to close stream for `ReusedExchange` operator. Follow up #1932. ### Why are the changes needed? `OpenStream` does not register chunk stream for reading local or dfs shuffle. Therefore `LocalPartitionReader` and `DfsPartitionReader` could not obtain the disk file from `ChunkStreamManager` that causes the below `NullPointerException` for closing stream. ``` ERROR [fetch-server-11-11] TransportRequestHandler: Error while invoking handler#receive() on RPC id 4 java.lang.NullPointerException at org.apache.celeborn.service.deploy.worker.storage.ChunkStreamManager.getShuffleKeyAndFileName(ChunkStreamManager.java:188) at org.apache.celeborn.service.deploy.worker.FetchHandler.handleEndStreamFromClient(FetchHandler.scala:344) at org.apache.celeborn.service.deploy.worker.FetchHandler.handleRpcRequest(FetchHandler.scala:137) at org.apache.celeborn.service.deploy.worker.FetchHandler.receive(FetchHandler.scala:94) at org.apache.celeborn.common.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:96) at org.apache.celeborn.common.network.server.TransportRequestHandler.handle(TransportRequestHandler.java:84) at org.apache.celeborn.common.network.server.TransportChannelHandler.channelRead(TransportChannelHandler.java:151) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) at org.apache.celeborn.common.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:74) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788) at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562) at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.lang.Thread.run(Thread.java:745) ``` In summary, `FetchHandler` only closes stream registered via `ChunkStreamManager`. `LocalPartitionReader` and `DfsPartitionReader` should use `ChunkStreamManager#registerStream` to close stream for deleting original unsorted disk file in `ReusedExchange` operator. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? - `FetchHandlerSuiteJ#testLocalReadSortFileOnceOriginalFileBeDeleted` - `FetchHandlerSuiteJ#testDoNotDeleteOriginalFileWhenNonRangeLocalReadWorkInProgress` - `ReusedExchangeSuite` Closes #2209 from SteNicholas/CELEBORN-1177. Authored-by: SteNicholas <programgeek@163.com> Signed-off-by: Fu Chen <cfmcgrady@gmail.com>

…ger to close stream for ReusedExchange `OpenStream` should register stream via `ChunkStreamManager`, which is served to obtain disk file to close stream for `ReusedExchange` operator. Follow up #1932. `OpenStream` does not register chunk stream for reading local or dfs shuffle. Therefore `LocalPartitionReader` and `DfsPartitionReader` could not obtain the disk file from `ChunkStreamManager` that causes the below `NullPointerException` for closing stream. ``` ERROR [fetch-server-11-11] TransportRequestHandler: Error while invoking handler#receive() on RPC id 4 java.lang.NullPointerException at org.apache.celeborn.service.deploy.worker.storage.ChunkStreamManager.getShuffleKeyAndFileName(ChunkStreamManager.java:188) at org.apache.celeborn.service.deploy.worker.FetchHandler.handleEndStreamFromClient(FetchHandler.scala:344) at org.apache.celeborn.service.deploy.worker.FetchHandler.handleRpcRequest(FetchHandler.scala:137) at org.apache.celeborn.service.deploy.worker.FetchHandler.receive(FetchHandler.scala:94) at org.apache.celeborn.common.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:96) at org.apache.celeborn.common.network.server.TransportRequestHandler.handle(TransportRequestHandler.java:84) at org.apache.celeborn.common.network.server.TransportChannelHandler.channelRead(TransportChannelHandler.java:151) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) at io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) at org.apache.celeborn.common.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:74) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788) at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562) at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) at java.lang.Thread.run(Thread.java:745) ``` In summary, `FetchHandler` only closes stream registered via `ChunkStreamManager`. `LocalPartitionReader` and `DfsPartitionReader` should use `ChunkStreamManager#registerStream` to close stream for deleting original unsorted disk file in `ReusedExchange` operator. No. - `FetchHandlerSuiteJ#testLocalReadSortFileOnceOriginalFileBeDeleted` - `FetchHandlerSuiteJ#testDoNotDeleteOriginalFileWhenNonRangeLocalReadWorkInProgress` - `ReusedExchangeSuite` Closes #2209 from SteNicholas/CELEBORN-1177. Authored-by: SteNicholas <programgeek@163.com> Signed-off-by: Fu Chen <cfmcgrady@gmail.com> (cherry picked from commit 7bc3497) Signed-off-by: Fu Chen <cfmcgrady@gmail.com>

cfmcgrady added 30 commits September 14, 2023 13:52

fix

c94e162

style

revert

59336f4

style

96a7aff

fix ci

a89d222

close dfs stream

8c3f242

fix ci

24f2c56

fix LocalReadWriteTestWithLZ4

89f16c4

migrate Flink client from using BufferStreamEnd to PbBufferStreamEnd

2dbe46e

import class

a5a630c

refactor

03e1a9d

style

61adc62

spark.stop

b4818b4

unnecessary change

318e1e4

refactor ReusedExchangeSuite

acfa117

simplify case

b06d28b

minor

6148252

style

ce1c8af

FetchHandler handle legacy rpc message

0bad21e

LocalPartitionReader closeStream

9af22ee

add testLegacyOpenStream

f86dc92

Merge remote-tracking branch 'apache/main' into fix-partition-sort-bu…

aea6943

…g-v4

Merge remote-tracking branch 'apache/main' into fix-partition-sort-bu…

4a13d13

…g-v4

ChunkStream/CreditStream

8d4dad0

todo

ff394f9

refactor

df87161

testLegacyOpenStream

a28931c

style

52995a0

nit

532f7ec

comment

dff73b9

style

1b6308b

cfmcgrady added 2 commits September 21, 2023 15:28

comment

f7bf186

comment

bedcb26

cfmcgrady mentioned this pull request Sep 21, 2023

[CELEBORN-980] Asynchronously delete original files to fix ReusedExchange bug #1907

Closed

FMX reviewed Sep 22, 2023

View reviewed changes

cfmcgrady added 6 commits September 22, 2023 12:36

Merge remote-tracking branch 'apache/main' into fix-partition-sort-bu…

3451146

…g-v4

resolve conflict

67010db

Merge remote-tracking branch 'apache/main' into fix-partition-sort-bu…

598a01d

…g-v4

style

d4f0bba

merge main

4092ce8

fix ci

988ca38

waitinfuture reviewed Oct 7, 2023

View reviewed changes

address comment

6c86e01

waitinfuture approved these changes Oct 9, 2023

View reviewed changes

waitinfuture closed this in c4135dc Oct 9, 2023

cfmcgrady mentioned this pull request Oct 18, 2023

[CELEBORN-1047] Remove conf celeborn.worker.sortPartition.eagerlyRemoveOriginalFiles.enabled #1999

Closed

cfmcgrady mentioned this pull request Oct 24, 2023

[CELEBORN-1082] Fixing partition sorter task failures due to duplicate sorting #2031

Closed

SteNicholas mentioned this pull request Jan 5, 2024

[CELEBORN-1177] OpenStream should register stream via ChunkStreamManager to close stream for ReusedExchange #2209

Closed

RexXiong mentioned this pull request Feb 12, 2025

[CELEBORN-1859] DfsPartitionReader and LocalPartitionReader should reuse pbStreamHandlers get from BatchOpenStream request #3093

Closed

Comments

Conversation

cfmcgrady commented Sep 21, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

codecov bot commented Sep 21, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

cfmcgrady commented Sep 21, 2023

Uh oh!

cfmcgrady commented Sep 21, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cfmcgrady Sep 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

waitinfuture Oct 7, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cfmcgrady commented Oct 9, 2023

Uh oh!

waitinfuture left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

FMX commented Oct 9, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

cfmcgrady commented Sep 21, 2023 •

edited

Loading

codecov bot commented Sep 21, 2023 •

edited

Loading

cfmcgrady Sep 22, 2023 •

edited

Loading

waitinfuture Oct 7, 2023 •

edited

Loading

waitinfuture left a comment •

edited

Loading