This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch refactor/orTimeout-futureTimeoutSupport in repository https://gitbox.apache.org/repos/asf/pekko.git
commit abb8fc766c95917b5c72ce439eccb74126e969b4 Author: 虎鸣 <[email protected]> AuthorDate: Tue Jun 23 09:30:07 2026 +0800 refactor: use CompletableFuture.orTimeout in FutureTimeoutSupport Motivation: timeoutCompletionStage was deprecated in favor of CompletableFuture.orTimeout but still used a manual Scheduler-based implementation internally. Modification: Rewrite timeoutCompletionStage to delegate to CompletableFuture.orTimeout(). Add a handle() wrapper to provide a consistent timeout message across JDK versions (JDK 25's orTimeout produces a null TimeoutException message, unlike JDK 17 which included a descriptive message). The Scheduler parameter is retained for API compatibility but no longer used. Result: Simpler implementation with ~5 fewer lines, no Scheduler dependency for timeout scheduling, and consistent timeout messages across JDK versions. Tests: sbt "actor-tests / Test / testOnly org.apache.pekko.pattern.PatternsTest" - 33/33 passed References: Refs #3136 --- .../pekko/pattern/FutureTimeoutSupport.scala | 27 +++++++++------------- 1 file changed, 11 insertions(+), 16 deletions(-) diff --git a/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala b/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala index 52709500f6..fe6e6117d7 100644 --- a/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala +++ b/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala @@ -134,23 +134,18 @@ trait FutureTimeoutSupport { catch { case NonFatal(t) => CompletableFuture.failedStage(t) } - if (stage.toCompletableFuture.isDone) { - stage - } else { - val p = new CompletableFuture[T] - val timeout = using.scheduleOnce(duration, - () => { - p.completeExceptionally(new TimeoutException(s"Timeout of $duration expired")) - stage.toCompletableFuture.cancel(true) - () - }) - stage.handle[Unit]((v: T, ex: Throwable) => { - timeout.cancel() - if (v != null) p.complete(v) - if (ex ne null) p.completeExceptionally(ex) + val millis = if (duration.isZero || duration.isNegative) 0L else duration.toMillis + val timeoutMessage = s"Timeout of $duration expired" + stage.toCompletableFuture + .orTimeout(millis, java.util.concurrent.TimeUnit.MILLISECONDS) + .handle((v: T, ex: Throwable) => { + if (ex ne null) { + ex match { + case _: TimeoutException => throw new TimeoutException(timeoutMessage) + case _ => throw ex + } + } else v }) - p - } } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
