chore: Make timeoutCompletionStage accept java Duration. (#2063) (#2071)

(cherry picked from commit 442ecd7)

Author: He-Pin

actor-tests/src/test/java/org/apache/pekko/pattern/PatternsTest.java

@@ -533,7 +533,7 @@ public void testCompletedWithTimeout() throws Exception {
       delayedStage.toCompletableFuture().get(3, SECONDS);
     } catch (ExecutionException e) {
       assertTrue(e.getCause() instanceof TimeoutException);
-      assertEquals("Timeout of 200 milliseconds expired", e.getCause().getMessage());
+      assertEquals("Timeout of PT0.2S expired", e.getCause().getMessage());
     }
   }
 

actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala

@@ -17,6 +17,7 @@ import java.util.concurrent.{ CompletableFuture, CompletionStage, TimeoutExcepti
 import scala.concurrent.{ ExecutionContext, Future, Promise }
 import scala.concurrent.duration.FiniteDuration
 import scala.util.control.NonFatal
+
 import org.apache.pekko
 import pekko.actor._
 import pekko.dispatch.Futures
@@ -126,7 +127,7 @@ trait FutureTimeoutSupport {
    * @since 1.2.0
    */
   @nowarn("msg=deprecated")
-  def timeoutCompletionStage[T](duration: FiniteDuration, using: Scheduler)(value: => CompletionStage[T])(
+  def timeoutCompletionStage[T](duration: java.time.Duration, using: Scheduler)(value: => CompletionStage[T])(
       implicit ec: ExecutionContext): CompletionStage[T] = {
     val stage: CompletionStage[T] =
       try value
@@ -137,10 +138,12 @@ trait FutureTimeoutSupport {
       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)
-      }
+      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)

actor/src/main/scala/org/apache/pekko/pattern/Patterns.scala

@@ -480,7 +480,7 @@ object Patterns {
       scheduler: Scheduler,
       context: ExecutionContext,
       value: Callable[CompletionStage[T]]): CompletionStage[T] =
-    timeoutCompletionStage(duration.asScala, scheduler)(value.call())(context)
+    timeoutCompletionStage(duration, scheduler)(value.call())(context)
 
   /**
    * Returns a [[scala.concurrent.Future]] that will be completed with the success or failure of the provided Callable