Use scala.concurrent.Batchable for Scala 2.13

akka-actor/src/main/scala-2.12/akka/dispatch/internal/ScalaBatchable.scala

@@ -0,0 +1,22 @@
+/*
+ * Copyright (C) 2009-2021 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.dispatch.internal
+
+import akka.annotation.InternalApi
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[akka] object ScalaBatchable {
+
+  // see Scala 2.13 source tree for explanation
+  def isBatchable(runnable: Runnable): Boolean = runnable match {
+    case b: akka.dispatch.Batchable             => b.isBatchable
+    case _: scala.concurrent.OnCompleteRunnable => true
+    case _                                      => false
+  }
+
+}

akka-actor/src/main/scala-2.13/akka/dispatch/internal/ScalaBatchable.scala

@@ -0,0 +1,28 @@
+/*
+ * Copyright (C) 2009-2021 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.dispatch.internal
+
+import akka.annotation.InternalApi
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[akka] object ScalaBatchable {
+
+  /*
+   * In Scala 2.13.0 `OnCompleteRunnable` was deprecated, and in 2.13.4
+   * `impl.Promise.Transformation` no longer extend the deprecated `OnCompleteRunnable`
+   * but instead `scala.concurrent.Batchable` (which anyway extends `OnCompleteRunnable`).
+   * On top of that we have or own legacy version `akka.dispatch.Batchable`.
+   */
+
+  def isBatchable(runnable: Runnable): Boolean = runnable match {
+    case b: akka.dispatch.Batchable    => b.isBatchable
+    case _: scala.concurrent.Batchable => true
+    case _                             => false
+  }
+
+}

akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala

@@ -38,11 +38,7 @@ object Envelope {
 }
 
 final case class TaskInvocation(eventStream: EventStream, runnable: Runnable, cleanup: () => Unit) extends Batchable {
-  final override def isBatchable: Boolean = runnable match {
-    case b: Batchable                           => b.isBatchable
-    case _: scala.concurrent.OnCompleteRunnable => true
-    case _                                      => false
-  }
+  final override def isBatchable: Boolean = akka.dispatch.internal.ScalaBatchable.isBatchable(runnable)
 
   def run(): Unit =
     try runnable.run()

akka-actor/src/main/scala/akka/dispatch/BatchingExecutor.scala

@@ -4,20 +4,26 @@
 
 package akka.dispatch
 
+import akka.annotation.InternalApi
+
 import java.util.ArrayDeque
 import java.util.concurrent.Executor
-
 import scala.annotation.tailrec
 import scala.concurrent._
 
 /**
+ * INTERNAL API
+ *
  * All Batchables are automatically batched when submitted to a BatchingExecutor
  */
+@InternalApi
 private[akka] trait Batchable extends Runnable {
   def isBatchable: Boolean
 }
 
 /**
+ * INTERNAL API
+ *
  * Mixin trait for an Executor
  * which groups multiple nested `Runnable.run()` calls
  * into a single Runnable passed to the original
@@ -45,6 +51,7 @@ private[akka] trait Batchable extends Runnable {
  * WARNING: The underlying Executor's execute-method must not execute the submitted Runnable
  * in the calling thread synchronously. It must enqueue/handoff the Runnable.
  */
+@InternalApi
 private[akka] trait BatchingExecutor extends Executor {
 
   // invariant: if "_tasksLocal.get ne null" then we are inside Batch.run; if it is null, we are outside
@@ -127,9 +134,5 @@ private[akka] trait BatchingExecutor extends Executor {
   }
 
   /** Override this to define which runnables will be batched. */
-  def batchable(runnable: Runnable): Boolean = runnable match {
-    case b: Batchable                           => b.isBatchable
-    case _: scala.concurrent.OnCompleteRunnable => true
-    case _                                      => false
-  }
+  def batchable(runnable: Runnable): Boolean = akka.dispatch.internal.ScalaBatchable.isBatchable(runnable)
 }