From cd7c98567ecbc564564a3f0b8dd6492cd351c237 Mon Sep 17 00:00:00 2001
From: PJ Fanning <pjfanning@users.noreply.github.com>
Date: Tue, 14 Jul 2026 13:06:58 +0100
Subject: [PATCH 1/6] use lambda instead of new Runnable

---
 .../ForkJoinPoolVirtualThreadSpec.scala       | 37 ++++++++-----------
 .../pekko/actor/dispatch/ActorModelSpec.scala | 11 +++---
 .../pekko/dispatch/ExecutionContextSpec.scala | 37 +++++++------------
 .../typed/internal/ActorSystemSpec.scala      |  4 +-
 .../org/apache/pekko/actor/ActorSystem.scala  |  2 +-
 .../actor/LightArrayRevolverScheduler.scala   |  4 +-
 .../org/apache/pekko/actor/Scheduler.scala    | 27 +++++---------
 .../dispatch/VirtualizedExecutorService.scala | 26 ++++++-------
 .../apache/pekko/pattern/CircuitBreaker.scala | 20 ++++------
 .../pekko/pattern/FutureTimeoutSupport.scala  | 22 +++++------
 .../OutputStreamSourceStageBenchmark.scala    | 14 +++----
 .../pekko/cluster/typed/ActorSystemSpec.scala |  4 +-
 .../org/apache/pekko/cluster/Cluster.scala    |  4 +-
 13 files changed, 87 insertions(+), 125 deletions(-)

diff --git a/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala b/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala
index 00f0fb6e0b9..a6138364eb4 100644
--- a/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala
+++ b/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala
@@ -130,12 +130,10 @@ class ForkJoinPoolVirtualThreadSpec extends PekkoSpec(ForkJoinPoolVirtualThreadS
       val finished = new CountDownLatch(1)
 
       try {
-        executor.execute(new Runnable {
-          override def run(): Unit = {
-            started.countDown()
-            release.await()
-            finished.countDown()
-          }
+        executor.execute(() => {
+          started.countDown()
+          release.await()
+          finished.countDown()
         })
 
         started.await(5, TimeUnit.SECONDS) should ===(true)
@@ -168,15 +166,14 @@ class ForkJoinPoolVirtualThreadSpec extends PekkoSpec(ForkJoinPoolVirtualThreadS
       val interrupted = new CountDownLatch(1)
 
       try {
-        executor.execute(new Runnable {
-          override def run(): Unit = {
-            started.countDown()
-            try release.await()
-            catch {
-              case _: InterruptedException => interrupted.countDown()
-            }
+        executor.execute(() => {
+          started.countDown()
+          try release.await()
+          catch {
+            case _: InterruptedException => interrupted.countDown()
           }
-        })
+        }
+        )
 
         started.await(5, TimeUnit.SECONDS) should ===(true)
         executor.shutdownNow()
@@ -205,13 +202,11 @@ class ForkJoinPoolVirtualThreadSpec extends PekkoSpec(ForkJoinPoolVirtualThreadS
       val interrupted = new CountDownLatch(1)
 
       try {
-        executor.execute(new Runnable {
-          override def run(): Unit = {
-            started.countDown()
-            try release.await()
-            catch {
-              case _: InterruptedException => interrupted.countDown()
-            }
+        executor.execute(() => {
+          started.countDown()
+          try release.await()
+          catch {
+            case _: InterruptedException => interrupted.countDown()
           }
         })
 
diff --git a/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/ActorModelSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/ActorModelSpec.scala
index 39bc418e69b..57449aa2c32 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/ActorModelSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/ActorModelSpec.scala
@@ -400,12 +400,11 @@ abstract class ActorModelSpec(config: String) extends PekkoSpec(config) with Def
           // this future is meant to keep the dispatcher alive until the end of the test run even if
           // the boss doesn't create children fast enough to keep the dispatcher from becoming empty
           // and it needs to be on a separate thread to not deadlock the calling thread dispatcher
-          new Thread(new Runnable {
-            def run() =
-              Future {
-                keepAliveLatch.await(waitTime, TimeUnit.MILLISECONDS)
-              }(dispatcher)
-          }).start()
+          new Thread(() =>
+            Future {
+              keepAliveLatch.await(waitTime, TimeUnit.MILLISECONDS)
+            }(dispatcher)
+          ).start()
           boss ! "run"
           assertCountDown(cachedMessage.latch, waitTime, "Counting down from " + num)
           assertCountDown(stopLatch, waitTime, "Expected all children to stop")
diff --git a/actor-tests/src/test/scala/org/apache/pekko/dispatch/ExecutionContextSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/dispatch/ExecutionContextSpec.scala
index 4db07018c29..d8db8583b81 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/dispatch/ExecutionContextSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/dispatch/ExecutionContextSpec.scala
@@ -107,20 +107,14 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
         val completed = new AtomicInteger(0)
         var innerBatch: Runnable = null
 
-        executor.execute(new Runnable {
-          override def run(): Unit = {
-            executor.execute(new Runnable {
-              override def run(): Unit = completed.incrementAndGet()
-            })
-            // Model a pool worker executing another submitted batch while helping a join.
-            innerBatch.run()
-            completed.incrementAndGet()
-          }
+        executor.execute(() => {
+          executor.execute(() => completed.incrementAndGet())
+          // Model a pool worker executing another submitted batch while helping a join.
+          innerBatch.run()
+          completed.incrementAndGet()
         })
         val outerBatch = submitted.remove()
-        executor.execute(new Runnable {
-          override def run(): Unit = completed.incrementAndGet()
-        })
+        executor.execute(() => completed.incrementAndGet())
         innerBatch = submitted.remove()
 
         outerBatch.run()
@@ -196,14 +190,11 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
     "work with same-thread executor plus blocking" in {
       val ec = scala.concurrent.ExecutionContext.parasitic
       var x = 0
-      ec.execute(new Runnable {
-        override def run = {
-          ec.execute(new Runnable {
-            override def run = blocking {
-              x = 1
-            }
+      ec.execute(() => {
+        ec.execute(() =>
+          blocking {
+            x = 1
           })
-        }
       })
       x should be(1)
     }
@@ -255,7 +246,7 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
     "be suspendable and resumable" in {
       val sec = SerializedSuspendableExecutionContext(1)(ExecutionContext.global)
       val counter = new AtomicInteger(0)
-      def perform(f: Int => Int) = sec.execute(new Runnable { def run = counter.set(f(counter.get)) })
+      def perform(f: Int => Int) = sec.execute(() => counter.set(f(counter.get)))
       perform(_ + 1)
       perform(x => { sec.suspend(); x * 2 })
       awaitCond(counter.get == 2)
@@ -282,7 +273,7 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
       val throughput = 25
       val sec = SerializedSuspendableExecutionContext(throughput)(underlying)
       sec.suspend()
-      def perform(f: Int => Int) = sec.execute(new Runnable { def run = counter.set(f(counter.get)) })
+      def perform(f: Int => Int) = sec.execute(() => counter.set(f(counter.get)))
 
       val total = 1000
       (1 to total).foreach { _ =>
@@ -299,7 +290,7 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
       val sec = SerializedSuspendableExecutionContext(1)(ExecutionContext.global)
       val total = 10000
       val counter = new AtomicInteger(0)
-      def perform(f: Int => Int) = sec.execute(new Runnable { def run = counter.set(f(counter.get)) })
+      def perform(f: Int => Int) = sec.execute(() => counter.set(f(counter.get)))
 
       (1 to total).foreach { i =>
         perform(c => if (c == (i - 1)) c + 1 else c)
@@ -318,7 +309,7 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D
       val throughput = 25
       val sec = SerializedSuspendableExecutionContext(throughput)(underlying)
       sec.suspend()
-      def perform(f: Int => Int) = sec.execute(new Runnable { def run = counter.set(f(counter.get)) })
+      def perform(f: Int => Int) = sec.execute(() => counter.set(f(counter.get)))
       perform(_ + 1)
       (1 to 10).foreach { _ =>
         perform(identity)
diff --git a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/internal/ActorSystemSpec.scala b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/internal/ActorSystemSpec.scala
index 094624d9674..a82b67b35b8 100644
--- a/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/internal/ActorSystemSpec.scala
+++ b/actor-typed-tests/src/test/scala/org/apache/pekko/actor/typed/internal/ActorSystemSpec.scala
@@ -156,9 +156,7 @@ class ActorSystemSpec
       withSystem("thread", Behaviors.empty[String]) { sys =>
         val p = Promise[Int]()
         sys.threadFactory
-          .newThread(new Runnable {
-            def run(): Unit = p.success(42)
-          })
+          .newThread(() => p.success(42))
           .start()
         p.future.futureValue should ===(42)
       }
diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala b/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
index 9c5accd751f..9e0a7c0b575 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
@@ -1102,7 +1102,7 @@ private[pekko] class ActorSystemImpl(
     }
 
   def start(): this.type = _start
-  def registerOnTermination[T](code: => T): Unit = { registerOnTermination(new Runnable { def run() = code }) }
+  def registerOnTermination[T](code: => T): Unit = { registerOnTermination(() => code) }
   def registerOnTermination(code: Runnable): Unit = { terminationCallbacks.add(code) }
 
   @volatile private var terminating = false
diff --git a/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala b/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
index 2d4e409501f..32261cb743e 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
@@ -417,8 +417,8 @@ object LightArrayRevolverScheduler {
     override def isCancelled: Boolean = task eq CancelledTask
   }
 
-  private val CancelledTask = new Runnable { def run = () }
-  private val ExecutedTask = new Runnable { def run = () }
+  private val CancelledTask = () => ()
+  private val ExecutedTask = () => ()
 
   private val NotCancellable: TimerTask = new TimerTask {
     def cancel(): Boolean = false
diff --git a/actor/src/main/scala/org/apache/pekko/actor/Scheduler.scala b/actor/src/main/scala/org/apache/pekko/actor/Scheduler.scala
index c6b36af79de..b7610314556 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/Scheduler.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/Scheduler.scala
@@ -156,12 +156,10 @@ trait Scheduler {
       implicit
       executor: ExecutionContext,
       sender: ActorRef = Actor.noSender): Cancellable = {
-    scheduleWithFixedDelay(initialDelay, delay)(new Runnable {
-      def run(): Unit = {
-        receiver ! message
-        if (receiver.isTerminated)
-          throw SchedulerException("timer active for terminated actor")
-      }
+    scheduleWithFixedDelay(initialDelay, delay)(() => {
+      receiver ! message
+      if (receiver.isTerminated)
+        throw SchedulerException("timer active for terminated actor")
     })
   }
 
@@ -301,12 +299,10 @@ trait Scheduler {
     schedule(
       initialDelay,
       interval,
-      new Runnable {
-        def run(): Unit = {
-          receiver ! message
-          if (receiver.isTerminated)
-            throw SchedulerException("timer active for terminated actor")
-        }
+      () => {
+        receiver ! message
+        if (receiver.isTerminated)
+          throw SchedulerException("timer active for terminated actor")
       })
 
   /**
@@ -360,10 +356,7 @@ trait Scheduler {
       implicit
       executor: ExecutionContext,
       sender: ActorRef = Actor.noSender): Cancellable =
-    scheduleOnce(delay,
-      new Runnable {
-        override def run(): Unit = receiver ! message
-      })
+    scheduleOnce(delay, () => receiver ! message)
 
   /**
    * Java API: Schedules a message to be sent once with a delay, i.e. a time period that has
@@ -396,7 +389,7 @@ trait Scheduler {
   final def scheduleOnce(delay: FiniteDuration)(f: => Unit)(
       implicit
       executor: ExecutionContext): Cancellable =
-    scheduleOnce(delay, new Runnable { override def run(): Unit = f })
+    scheduleOnce(delay, () => f)
 
   /**
    * Scala API: Schedules a Runnable to be run once with a delay, i.e. a time period that
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/VirtualizedExecutorService.scala b/actor/src/main/scala/org/apache/pekko/dispatch/VirtualizedExecutorService.scala
index 160224cb047..9fac9278f0b 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/VirtualizedExecutorService.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/VirtualizedExecutorService.scala
@@ -100,22 +100,20 @@ final class VirtualizedExecutorService(
       } else if (underlyingShutdownScheduled.compareAndSet(false, true)) {
         VirtualThreadSupport.startVirtualThread(
           "pekko-virtualized-executor-shutdown",
-          new Runnable {
-            override def run(): Unit = {
-              var interrupted = false
-              try {
-                while (!executor.isTerminated) {
-                  try executor.awaitTermination(Long.MaxValue, TimeUnit.NANOSECONDS)
-                  catch {
-                    case _: InterruptedException => interrupted = true
-                  }
-                }
-              } finally {
-                shutdownUnderlying()
-                if (interrupted) {
-                  Thread.currentThread().interrupt()
+          () => {
+            var interrupted = false
+            try {
+              while (!executor.isTerminated) {
+                try executor.awaitTermination(Long.MaxValue, TimeUnit.NANOSECONDS)
+                catch {
+                  case _: InterruptedException => interrupted = true
                 }
               }
+            } finally {
+              shutdownUnderlying()
+              if (interrupted) {
+                Thread.currentThread().interrupt()
+              }
             }
           })
       }
diff --git a/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala b/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala
index 2a7b8b7b520..0e4cba4bcf0 100644
--- a/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala
+++ b/actor/src/main/scala/org/apache/pekko/pattern/CircuitBreaker.scala
@@ -497,7 +497,7 @@ class CircuitBreaker(
    * @param callback Handler to be invoked on state change
    * @return CircuitBreaker for fluent usage
    */
-  def onOpen(callback: => Unit): CircuitBreaker = addOnOpenListener(new Runnable { def run = callback })
+  def onOpen(callback: => Unit): CircuitBreaker = addOnOpenListener(() => callback)
 
   /**
    * Java API for onOpen
@@ -517,7 +517,7 @@ class CircuitBreaker(
    * @param callback Handler to be invoked on state change
    * @return CircuitBreaker for fluent usage
    */
-  def onHalfOpen(callback: => Unit): CircuitBreaker = addOnHalfOpenListener(new Runnable { def run = callback })
+  def onHalfOpen(callback: => Unit): CircuitBreaker = addOnHalfOpenListener(() => callback)
 
   /**
    * JavaAPI for onHalfOpen
@@ -538,7 +538,7 @@ class CircuitBreaker(
    * @param callback Handler to be invoked on state change
    * @return CircuitBreaker for fluent usage
    */
-  def onClose(callback: => Unit): CircuitBreaker = addOnCloseListener(new Runnable { def run = callback })
+  def onClose(callback: => Unit): CircuitBreaker = addOnCloseListener(() => callback)
 
   /**
    * JavaAPI for onClose
@@ -632,7 +632,7 @@ class CircuitBreaker(
    * @return CircuitBreaker for fluent usage
    */
   def onCallBreakerOpen(callback: => Unit): CircuitBreaker =
-    addOnCallBreakerOpenListener(new Runnable { def run = callback })
+    addOnCallBreakerOpenListener(() => callback)
 
   /**
    * JavaAPI for onCallBreakerOpen.
@@ -686,9 +686,7 @@ class CircuitBreaker(
     val iterator = successListeners.iterator()
     while (iterator.hasNext) {
       val listener = iterator.next()
-      executor.execute(new Runnable {
-        def run() = listener.accept(elapsed)
-      })
+      executor.execute(() => listener.accept(elapsed))
     }
   }
 
@@ -702,9 +700,7 @@ class CircuitBreaker(
     val iterator = callFailureListeners.iterator()
     while (iterator.hasNext) {
       val listener = iterator.next()
-      executor.execute(new Runnable {
-        def run() = listener.accept(elapsed)
-      })
+      executor.execute(() => listener.accept(elapsed))
     }
   }
 
@@ -718,9 +714,7 @@ class CircuitBreaker(
     val iterator = callTimeoutListeners.iterator()
     while (iterator.hasNext) {
       val listener = iterator.next()
-      executor.execute(new Runnable {
-        def run() = listener.accept(elapsed)
-      })
+      executor.execute(() => listener.accept(elapsed))
     }
   }
 
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 3e696758f92..1c76873333c 100644
--- a/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala
+++ b/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala
@@ -74,18 +74,16 @@ trait FutureTimeoutSupport {
       val p = new CompletableFuture[T]
       using.scheduleOnce(
         duration,
-        new Runnable {
-          override def run(): Unit =
-            try {
-              val future = value
-              future.handle[Unit]((t: T, ex: Throwable) => {
-                if (t != null) p.complete(t)
-                if (ex ne null) p.completeExceptionally(ex)
-              })
-            } catch {
-              case NonFatal(ex) => p.completeExceptionally(ex)
-            }
-        })
+        () =>
+          try {
+            val future = value
+            future.handle[Unit]((t: T, ex: Throwable) => {
+              if (t != null) p.complete(t)
+              if (ex ne null) p.completeExceptionally(ex)
+            })
+          } catch {
+            case NonFatal(ex) => p.completeExceptionally(ex)
+          })
       p
     }
 
diff --git a/bench-jmh/src/main/scala/org/apache/pekko/stream/impl/OutputStreamSourceStageBenchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/stream/impl/OutputStreamSourceStageBenchmark.scala
index 4b06c731d79..ecf21f386ff 100644
--- a/bench-jmh/src/main/scala/org/apache/pekko/stream/impl/OutputStreamSourceStageBenchmark.scala
+++ b/bench-jmh/src/main/scala/org/apache/pekko/stream/impl/OutputStreamSourceStageBenchmark.scala
@@ -43,15 +43,13 @@ class OutputStreamSourceStageBenchmark {
   @OperationsPerInvocation(WritesPerBench)
   def consumeWrites(): Unit = {
     val (os, done) = StreamConverters.asOutputStream().toMat(Sink.ignore)(Keep.both).run()
-    new Thread(new Runnable {
-      def run(): Unit = {
-        var counter = 0
-        while (counter > WritesPerBench) {
-          os.write(bytes)
-          counter += 1
-        }
-        os.close()
+    new Thread(() => {
+      var counter = 0
+      while (counter > WritesPerBench) {
+        os.write(bytes)
+        counter += 1
       }
+      os.close()
     }).start()
     Await.result(done, 30.seconds)
   }
diff --git a/cluster-typed/src/test/scala/org/apache/pekko/cluster/typed/ActorSystemSpec.scala b/cluster-typed/src/test/scala/org/apache/pekko/cluster/typed/ActorSystemSpec.scala
index ae5e9d66dab..642f192da21 100644
--- a/cluster-typed/src/test/scala/org/apache/pekko/cluster/typed/ActorSystemSpec.scala
+++ b/cluster-typed/src/test/scala/org/apache/pekko/cluster/typed/ActorSystemSpec.scala
@@ -205,9 +205,7 @@ class ActorSystemSpec
       withSystem("thread", Behaviors.empty[String]) { sys =>
         val p = Promise[Int]()
         sys.threadFactory
-          .newThread(new Runnable {
-            def run(): Unit = p.success(42)
-          })
+          .newThread(() => p.success(42))
           .start()
         p.future.futureValue should ===(42)
       }
diff --git a/cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala b/cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala
index 4555dca23ed..490574288f4 100644
--- a/cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala
+++ b/cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala
@@ -439,7 +439,7 @@ class Cluster(val system: ExtendedActorSystem) extends Extension {
    * a certain size.
    */
   def registerOnMemberUp[T](code: => T): Unit =
-    registerOnMemberUp(new Runnable { def run() = code })
+    registerOnMemberUp(() => code)
 
   /**
    * Java API: The supplied callback will be run, once, when current cluster member is `Up`.
@@ -457,7 +457,7 @@ class Cluster(val system: ExtendedActorSystem) extends Extension {
    * is not invoked. It's often better to use [[pekko.actor.CoordinatedShutdown]] for this purpose.
    */
   def registerOnMemberRemoved[T](code: => T): Unit =
-    registerOnMemberRemoved(new Runnable { override def run(): Unit = code })
+    registerOnMemberRemoved(() => code)
 
   /**
    * Java API: The supplied thunk will be run, once, when current cluster member is `Removed`.

From f91626e823c0604131f1720b6f2411436e21a925 Mon Sep 17 00:00:00 2001
From: PJ Fanning <pjfanning@users.noreply.github.com>
Date: Tue, 14 Jul 2026 13:09:19 +0100
Subject: [PATCH 2/6] more

---
 .../pekko/stream/scaladsl/FlowConcatSpec.scala       | 12 +++---------
 .../pekko/stream/scaladsl/StreamConverters.scala     |  2 +-
 .../scala/org/apache/pekko/testkit/Coroner.scala     |  2 +-
 3 files changed, 5 insertions(+), 11 deletions(-)

diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowConcatSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowConcatSpec.scala
index e4aa94abe67..06bfd26831f 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowConcatSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowConcatSpec.scala
@@ -303,9 +303,7 @@ abstract class AbstractFlowConcatSpec extends BaseTwoStreamsSetup {
       val closed = new AtomicBoolean(false)
       val s1 = Source.single(1)
       val s2 = Source.fromJavaStream { () =>
-        Collections.singleton(2: Integer).stream().onClose(new Runnable {
-          override def run(): Unit = closed.set(true)
-        })
+        Collections.singleton(2: Integer).stream().onClose(() => closed.set(true))
       }
       val concat = if (eager) s1.concat(s2) else s1.concatLazy(s2)
       if (!eager) concat.traversalBuilder.pendingBuilder.toString should include("JavaStreamConcat")
@@ -320,9 +318,7 @@ abstract class AbstractFlowConcatSpec extends BaseTwoStreamsSetup {
       val s1 = Source.single(1)
       val s2 = Source.fromJavaStream { () =>
         java.util.stream.IntStream.rangeClosed(2, 1000).boxed().asInstanceOf[java.util.stream.Stream[Integer]]
-          .onClose(new Runnable {
-            override def run(): Unit = closed.set(true)
-          })
+          .onClose(() => closed.set(true))
       }
       val concat = if (eager) s1.concat(s2) else s1.concatLazy(s2)
       if (!eager) concat.traversalBuilder.pendingBuilder.toString should include("JavaStreamConcat")
@@ -344,9 +340,7 @@ abstract class AbstractFlowConcatSpec extends BaseTwoStreamsSetup {
         // Build a stream whose iterator.next() throws on the first call.
         java.util.stream.Stream.of[Integer](2: Integer).map[Integer](new java.util.function.Function[Integer, Integer] {
           override def apply(t: Integer): Integer = throw ex
-        }).onClose(new Runnable {
-          override def run(): Unit = closed.set(true)
-        })
+        }).onClose(() => closed.set(true))
       }
       val concat = if (eager) s1.concat(s2) else s1.concatLazy(s2)
       if (!eager) concat.traversalBuilder.pendingBuilder.toString should include("JavaStreamConcat")
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/StreamConverters.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/StreamConverters.scala
index d010f59b46a..d35a03f60ad 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/StreamConverters.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/StreamConverters.scala
@@ -209,7 +209,7 @@ object StreamConverters {
               },
               0),
             false)
-          .onClose(new Runnable { def run = queue.cancel() }))
+          .onClose(() => queue.cancel()))
       .withAttributes(DefaultAttributes.asJavaStream)
   }
 
diff --git a/testkit/src/test/scala/org/apache/pekko/testkit/Coroner.scala b/testkit/src/test/scala/org/apache/pekko/testkit/Coroner.scala
index 2b3c1d50e27..c03b57e46e1 100644
--- a/testkit/src/test/scala/org/apache/pekko/testkit/Coroner.scala
+++ b/testkit/src/test/scala/org/apache/pekko/testkit/Coroner.scala
@@ -128,7 +128,7 @@ object Coroner {
         watchedHandle.finished()
       }
     }
-    new Thread(new Runnable { def run = triggerReportIfOverdue(duration) }, "Coroner").start()
+    new Thread(() => triggerReportIfOverdue(duration), "Coroner").start()
     watchedHandle.waitForStart()
     watchedHandle
   }

From a0fe064c1dc45b2d5142b2c45ae8d0732de041d2 Mon Sep 17 00:00:00 2001
From: PJ Fanning <pjfanning@users.noreply.github.com>
Date: Tue, 14 Jul 2026 13:25:38 +0100
Subject: [PATCH 3/6] revert some changes

---
 .../actor/LightArrayRevolverScheduler.scala   |  4 ++--
 .../pekko/pattern/FutureTimeoutSupport.scala  | 22 ++++++++++---------
 2 files changed, 14 insertions(+), 12 deletions(-)

diff --git a/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala b/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
index 32261cb743e..2fe787cb1e7 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/LightArrayRevolverScheduler.scala
@@ -417,8 +417,8 @@ object LightArrayRevolverScheduler {
     override def isCancelled: Boolean = task eq CancelledTask
   }
 
-  private val CancelledTask = () => ()
-  private val ExecutedTask = () => ()
+  private val CancelledTask: Runnable = () => ()
+  private val ExecutedTask: Runnable = () => ()
 
   private val NotCancellable: TimerTask = new TimerTask {
     def cancel(): Boolean = false
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 1c76873333c..3e696758f92 100644
--- a/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala
+++ b/actor/src/main/scala/org/apache/pekko/pattern/FutureTimeoutSupport.scala
@@ -74,16 +74,18 @@ trait FutureTimeoutSupport {
       val p = new CompletableFuture[T]
       using.scheduleOnce(
         duration,
-        () =>
-          try {
-            val future = value
-            future.handle[Unit]((t: T, ex: Throwable) => {
-              if (t != null) p.complete(t)
-              if (ex ne null) p.completeExceptionally(ex)
-            })
-          } catch {
-            case NonFatal(ex) => p.completeExceptionally(ex)
-          })
+        new Runnable {
+          override def run(): Unit =
+            try {
+              val future = value
+              future.handle[Unit]((t: T, ex: Throwable) => {
+                if (t != null) p.complete(t)
+                if (ex ne null) p.completeExceptionally(ex)
+              })
+            } catch {
+              case NonFatal(ex) => p.completeExceptionally(ex)
+            }
+        })
       p
     }
 

From 4dc60f25ff30f34b3d6f74be3f7a6c6b7cdc9820 Mon Sep 17 00:00:00 2001
From: PJ Fanning <pjfanning@users.noreply.github.com>
Date: Tue, 14 Jul 2026 13:54:38 +0100
Subject: [PATCH 4/6] Update ActorModelSpec.scala

---
 .../apache/pekko/actor/dispatch/ActorModelSpec.scala  | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/ActorModelSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/ActorModelSpec.scala
index 57449aa2c32..39bc418e69b 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/ActorModelSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/dispatch/ActorModelSpec.scala
@@ -400,11 +400,12 @@ abstract class ActorModelSpec(config: String) extends PekkoSpec(config) with Def
           // this future is meant to keep the dispatcher alive until the end of the test run even if
           // the boss doesn't create children fast enough to keep the dispatcher from becoming empty
           // and it needs to be on a separate thread to not deadlock the calling thread dispatcher
-          new Thread(() =>
-            Future {
-              keepAliveLatch.await(waitTime, TimeUnit.MILLISECONDS)
-            }(dispatcher)
-          ).start()
+          new Thread(new Runnable {
+            def run() =
+              Future {
+                keepAliveLatch.await(waitTime, TimeUnit.MILLISECONDS)
+              }(dispatcher)
+          }).start()
           boss ! "run"
           assertCountDown(cachedMessage.latch, waitTime, "Counting down from " + num)
           assertCountDown(stopLatch, waitTime, "Expected all children to stop")

From 908aa4d3ef0ae147c362e8eb39c2237f0c77fa2a Mon Sep 17 00:00:00 2001
From: PJ Fanning <pjfanning@users.noreply.github.com>
Date: Tue, 14 Jul 2026 14:10:39 +0100
Subject: [PATCH 5/6] Update ActorSystem.scala

---
 actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala b/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
index 9e0a7c0b575..1536fbdf06a 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/ActorSystem.scala
@@ -1102,7 +1102,7 @@ private[pekko] class ActorSystemImpl(
     }
 
   def start(): this.type = _start
-  def registerOnTermination[T](code: => T): Unit = { registerOnTermination(() => code) }
+  def registerOnTermination[T](code: => T): Unit = { registerOnTermination((() => code): Runnable) }
   def registerOnTermination(code: Runnable): Unit = { terminationCallbacks.add(code) }
 
   @volatile private var terminating = false

From 175a1e3d2d2a0ed7c61610b62014456fc7603301 Mon Sep 17 00:00:00 2001
From: PJ Fanning <pjfanning@users.noreply.github.com>
Date: Tue, 14 Jul 2026 15:15:09 +0100
Subject: [PATCH 6/6] Update Cluster.scala

---
 cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala b/cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala
index 490574288f4..a96b5f2eee6 100644
--- a/cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala
+++ b/cluster/src/main/scala/org/apache/pekko/cluster/Cluster.scala
@@ -439,7 +439,7 @@ class Cluster(val system: ExtendedActorSystem) extends Extension {
    * a certain size.
    */
   def registerOnMemberUp[T](code: => T): Unit =
-    registerOnMemberUp(() => code)
+    registerOnMemberUp((() => code): Runnable)
 
   /**
    * Java API: The supplied callback will be run, once, when current cluster member is `Up`.
@@ -457,7 +457,7 @@ class Cluster(val system: ExtendedActorSystem) extends Extension {
    * is not invoked. It's often better to use [[pekko.actor.CoordinatedShutdown]] for this purpose.
    */
   def registerOnMemberRemoved[T](code: => T): Unit =
-    registerOnMemberRemoved(() => code)
+    registerOnMemberRemoved((() => code): Runnable)
 
   /**
    * Java API: The supplied thunk will be run, once, when current cluster member is `Removed`.