fix

apache · Oct 11, 2024 · e7e2883 · e7e2883
1 parent 7c8e9dd
commit e7e2883
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 4 deletions.
diff --git a/...n/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java b/...n/java/org/apache/celeborn/service/deploy/worker/congestcontrol/CongestionController.java
@@ -157,9 +157,9 @@ public boolean isUserCongested(UserCongestionControlContext userCongestionContro
     }
 
     UserIdentifier userIdentifier = userCongestionControlContext.getUserIdentifier();
+    long userProduceSpeed = getUserProduceSpeed(userCongestionControlContext.getUserBufferInfo());
     // If the user produce speed is higher that the avg consume speed, will congest it
     if (overHighWatermark.get()) {
-      long userProduceSpeed = getUserProduceSpeed(userCongestionControlContext.getUserBufferInfo());
       long avgConsumeSpeed = getPotentialProduceSpeed();
       if (userProduceSpeed > avgConsumeSpeed) {
         if (logger.isDebugEnabled()) {
@@ -173,7 +173,6 @@ public boolean isUserCongested(UserCongestionControlContext userCongestionContro
       }
     }
 
-    long userProduceSpeed = getUserProduceSpeed(userCongestionControlContext.getUserBufferInfo());
     if (userProduceSpeed > userProduceSpeedHighWatermark) {
       userCongestionControlContext.onCongestionControl();
       if (logger.isDebugEnabled()) {

diff --git a/...rg/apache/celeborn/service/deploy/worker/congestcontrol/UserCongestionControlContext.java b/...rg/apache/celeborn/service/deploy/worker/congestcontrol/UserCongestionControlContext.java
@@ -61,9 +61,10 @@ public boolean inCongestionControl() {
   }
 
   public void updateProduceBytes(long numBytes) {
+    long timeNow = System.currentTimeMillis();
     BufferStatusHub.BufferStatusNode node = new BufferStatusHub.BufferStatusNode(numBytes);
-    userBufferInfo.updateInfo(System.currentTimeMillis(), node);
-    workerBufferStatusHub.add(System.currentTimeMillis(), node);
+    userBufferInfo.updateInfo(timeNow, node);
+    workerBufferStatusHub.add(timeNow, node);
   }
 
   public UserBufferInfo getUserBufferInfo() {