diff --git a/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala b/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala index 15db160d9cb..5f7f6b110b7 100644 --- a/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala +++ b/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala @@ -23,7 +23,6 @@ import java.util.concurrent.{Callable, ConcurrentHashMap, ThreadPoolExecutor, Ti import scala.collection.JavaConverters._ import scala.collection.mutable -import scala.util.Random import com.google.common.cache.{Cache, CacheBuilder} @@ -303,7 +302,6 @@ class ReducePartitionCommitHandler( context.asInstanceOf[RemoteNettyRpcCallContext].nettyEnv.serialize(returnedMsg) } }) - context.asInstanceOf[RemoteNettyRpcCallContext].callback.onSuccess(cachedMsg) } } diff --git a/client/src/test/java/org/apache/celeborn/client/ShuffleClientDelayedResponseSuiteJ.java b/client/src/test/java/org/apache/celeborn/client/ShuffleClientDelayedResponseSuiteJ.java index 6ad67df3887..173395d2b3c 100644 --- a/client/src/test/java/org/apache/celeborn/client/ShuffleClientDelayedResponseSuiteJ.java +++ b/client/src/test/java/org/apache/celeborn/client/ShuffleClientDelayedResponseSuiteJ.java @@ -59,7 +59,7 @@ public void testUpdateReducerFileGroupTimeout() throws InterruptedException { when(endpointRef.askSync(any(), any(), any())) .thenAnswer( new AnswersWithDelay( - 200, + 2000, invocation -> GetReducerFileGroupResponse$.MODULE$.apply( StatusCode.SUCCESS, locations, new int[0], Collections.emptySet()))); @@ -76,7 +76,7 @@ public void testUpdateReducerFileGroupTimeout() throws InterruptedException { exceptionRef.set(e); } - Thread.sleep(5000); + Thread.sleep(1000); Exception exception = exceptionRef.get(); Assert.assertTrue(exception.getCause() instanceof TimeoutException);