This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 221129bb85 [CELEBORN-2343] Fix timer leak in handlePushData
221129bb85 is described below
commit 221129bb858f14743412a1d7a25886540333d94e
Author: Sanskar Modi <[email protected]>
AuthorDate: Wed Jun 3 10:48:22 2026 +0800
[CELEBORN-2343] Fix timer leak in handlePushData
### What changes were proposed in this pull request?
Fix the timer leak in handlePushData
### Why are the changes needed?
To avoid timer leak and publish correct `PRIMARY_PUSH_DATA_TIME` and
`REPLICA_PUSH_DATA_TIME` metrics
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
NA
Closes #3709 from s0nskar/fix_callback.
Authored-by: Sanskar Modi <[email protected]>
Signed-off-by: SteNicholas <[email protected]>
---
.../org/apache/celeborn/service/deploy/worker/PushDataHandler.scala | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
index cc1dc257f2..0accb33adb 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala
@@ -257,7 +257,7 @@ class PushDataHandler(val workerSource: WorkerSource)
extends BaseMessageHandler
val splitStatus = checkDiskFullAndSplit(fileWriter, isPrimary)
if (splitStatus == StatusCode.HARD_SPLIT) {
workerSource.incCounter(WorkerSource.WRITE_DATA_HARD_SPLIT_COUNT)
-
callback.onSuccess(ByteBuffer.wrap(Array[Byte](StatusCode.HARD_SPLIT.getValue)))
+
callbackWithTimer.onSuccess(ByteBuffer.wrap(Array[Byte](StatusCode.HARD_SPLIT.getValue)))
return
} else if (splitStatus == StatusCode.SOFT_SPLIT) {
softSplit = true
@@ -304,7 +304,7 @@ class PushDataHandler(val workerSource: WorkerSource)
extends BaseMessageHandler
Try(Await.result(writePromise.future, Duration.Inf)) match {
case Success(result) =>
if (result(0) != StatusCode.SUCCESS) {
-
callback.onSuccess(ByteBuffer.wrap(Array[Byte](result(0).getValue)))
+
callbackWithTimer.onSuccess(ByteBuffer.wrap(Array[Byte](result(0).getValue)))
} else {
if (response.remaining() > 0) {
val resp = ByteBuffer.allocate(response.remaining())
@@ -399,7 +399,7 @@ class PushDataHandler(val workerSource: WorkerSource)
extends BaseMessageHandler
Try(Await.result(writePromise.future, Duration.Inf)) match {
case Success(result) =>
if (result(0) != StatusCode.SUCCESS) {
-
callback.onSuccess(ByteBuffer.wrap(Array[Byte](result(0).getValue)))
+
callbackWithTimer.onSuccess(ByteBuffer.wrap(Array[Byte](result(0).getValue)))
} else {
if (softSplit) {
callbackWithTimer.onSuccess(
