apache/celeborn
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[CELEBORN-2033] updateProduceBytes should be called even if updateProduceBytes throws exception

Browse Source 
### What changes were proposed in this pull request?
updateProduceBytes should be called even if updateProduceBytes throws exception

### Why are the changes needed?
To make UserProduceSpeed ​​metrics more accurate

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Existing UTs.

Closes #3322 from leixm/CELEBORN-2033.

Authored-by: Xianming Lei <xianming.lei@shopee.com>
Signed-off-by: Wang, Fei <fwang12@ebay.com>
This commit is contained in:
Xianming Lei 2025-06-11 10:54:24 -07:00 committed by Wang, Fei
parent edeeb4b30a
commit bbd3bb4814
1 changed files with 2 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala
View File

@ -412,6 +412,8 @@ class LocalTierWriter(
override def writeInternal(buf: ByteBuf): Unit = {
val numBytes = buf.readableBytes()
if (userCongestionControlContext != null)
userCongestionControlContext.updateProduceBytes(numBytes)
val flushBufferReadableBytes = flushBuffer.readableBytes
if (flushBufferReadableBytes != 0 && flushBufferReadableBytes + numBytes >= flusherBufferSize) {
flush(false)
@ -422,13 +424,9 @@ class LocalTierWriter(
} catch {
case oom: OutOfMemoryError =>
MemoryManager.instance.incrementDiskBuffer(numBytes)
if (userCongestionControlContext != null)
userCongestionControlContext.updateProduceBytes(numBytes)
throw oom
}
MemoryManager.instance.incrementDiskBuffer(numBytes)
if (userCongestionControlContext != null)
userCongestionControlContext.updateProduceBytes(numBytes)
}
override def evict(file: TierWriterBase): Unit = ???