Typed Distributed Data requires untyped Cluster #25746 (#26074)

Typed Distributed Data requires untyped Cluster [#25746](https://github.com/akka/akka/issues/25746)

akka-distributed-data/src/main/scala/akka/cluster/ddata/DistributedData.scala

@@ -10,7 +10,7 @@ import akka.actor.ExtendedActorSystem
 import akka.actor.Extension
 import akka.actor.ExtensionId
 import akka.actor.ExtensionIdProvider
-import akka.cluster.Cluster
+import akka.cluster.{ Cluster, UniqueAddress }
 
 object DistributedData extends ExtensionId[DistributedData] with ExtensionIdProvider {
   override def get(system: ActorSystem): DistributedData = super.get(system)
@@ -28,14 +28,9 @@ object DistributedData extends ExtensionId[DistributedData] with ExtensionIdProv
  */
 class DistributedData(system: ExtendedActorSystem) extends Extension {
 
-  private val config = system.settings.config.getConfig("akka.cluster.distributed-data")
-  private val settings = ReplicatorSettings(config)
+  private val settings = ReplicatorSettings(system)
 
-  /**
-   * Returns true if this member is not tagged with the role configured for the
-   * replicas.
-   */
-  def isTerminated: Boolean = Cluster(system).isTerminated || !settings.roles.subsetOf(Cluster(system).selfRoles)
+  implicit val selfUniqueAddress: SelfUniqueAddress = SelfUniqueAddress(Cluster(system).selfUniqueAddress)
 
   /**
    * `ActorRef` of the [[Replicator]] .
@@ -45,7 +40,20 @@ class DistributedData(system: ExtendedActorSystem) extends Extension {
       system.log.warning("Replicator points to dead letters: Make sure the cluster node is not terminated and has the proper role!")
       system.deadLetters
     } else {
-      val name = config.getString("name")
-      system.systemActorOf(Replicator.props(settings), name)
+      system.systemActorOf(Replicator.props(settings), ReplicatorSettings.name(system, None))
     }
+
+  /**
+   * Returns true if this member is not tagged with the role configured for the
+   * replicas.
+   */
+  def isTerminated: Boolean =
+    Cluster(system).isTerminated || !settings.roles.subsetOf(Cluster(system).selfRoles)
+
 }
+
+/**
+ * Cluster non-specific (typed vs untyped) wrapper for [[akka.cluster.UniqueAddress]].
+ */
+@SerialVersionUID(1L)
+final case class SelfUniqueAddress(uniqueAddress: UniqueAddress)

akka-distributed-data/src/main/scala/akka/cluster/ddata/GCounter.scala

@@ -65,14 +65,19 @@ final class GCounter private[akka] (
    * Increment the counter with the delta `n` specified.
    * The delta must be zero or positive.
    */
-  def +(n: Long)(implicit node: Cluster): GCounter = increment(node, n)
+  def :+(n: Long)(implicit node: SelfUniqueAddress): GCounter = increment(node.uniqueAddress, n)
+
+  @deprecated("Use `:+` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def +(n: Long)(implicit node: Cluster): GCounter = increment(node.selfUniqueAddress, n)
 
   /**
    * Increment the counter with the delta `n` specified.
    * The delta `n` must be zero or positive.
    */
-  def increment(node: Cluster, n: Long = 1): GCounter =
-    increment(node.selfUniqueAddress, n)
+  def increment(node: SelfUniqueAddress, n: Long): GCounter = increment(node.uniqueAddress, n)
+
+  @deprecated("Use `increment` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def increment(node: Cluster, n: Long = 1): GCounter = increment(node.selfUniqueAddress, n)
 
   /**
    * INTERNAL API

akka-distributed-data/src/main/scala/akka/cluster/ddata/LWWMap.scala

@@ -4,9 +4,9 @@
 
 package akka.cluster.ddata
 
+import akka.annotation.InternalApi
 import akka.cluster.Cluster
 import akka.cluster.UniqueAddress
-import akka.annotation.InternalApi
 import akka.cluster.ddata.ORMap.ZeroTag
 
 object LWWMap {
@@ -87,6 +87,12 @@ final class LWWMap[A, B] private[akka] (
   /**
    * Adds an entry to the map
    */
+  def :+(entry: (A, B))(implicit node: SelfUniqueAddress): LWWMap[A, B] = {
+    val (key, value) = entry
+    put(node, key, value)
+  }
+
+  @deprecated("Use `:+` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def +(entry: (A, B))(implicit node: Cluster): LWWMap[A, B] = {
     val (key, value) = entry
     put(node, key, value)
@@ -95,8 +101,12 @@ final class LWWMap[A, B] private[akka] (
   /**
    * Adds an entry to the map
    */
+  def put(node: SelfUniqueAddress, key: A, value: B): LWWMap[A, B] =
+    put(node.uniqueAddress, key, value, defaultClock[B])
+
+  @deprecated("Use `put` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def put(node: Cluster, key: A, value: B): LWWMap[A, B] =
-    put(node, key, value, defaultClock[B])
+    put(node.selfUniqueAddress, key, value, defaultClock[B])
 
   /**
    * Adds an entry to the map.
@@ -106,6 +116,10 @@ final class LWWMap[A, B] private[akka] (
    * increasing version number from a database record that is used for optimistic
    * concurrency control.
    */
+  def put(node: SelfUniqueAddress, key: A, value: B, clock: Clock[B]): LWWMap[A, B] =
+    put(node.uniqueAddress, key, value, clock)
+
+  @deprecated("Use `put` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def put(node: Cluster, key: A, value: B, clock: Clock[B]): LWWMap[A, B] =
     put(node.selfUniqueAddress, key, value, clock)
 
@@ -117,8 +131,9 @@ final class LWWMap[A, B] private[akka] (
    * increasing version number from a database record that is used for optimistic
    * concurrency control.
    */
+  @deprecated("Use `put` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def put(key: A, value: B)(implicit node: Cluster, clock: Clock[B] = defaultClock[B]): LWWMap[A, B] =
-    put(node, key, value, clock)
+    put(node.selfUniqueAddress, key, value, clock)
 
   /**
    * INTERNAL API
@@ -136,6 +151,7 @@ final class LWWMap[A, B] private[akka] (
    * Note that if there is a conflicting update on another node the entry will
    * not be removed after merge.
    */
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def -(key: A)(implicit node: Cluster): LWWMap[A, B] = remove(node, key)
 
   /**
@@ -143,6 +159,10 @@ final class LWWMap[A, B] private[akka] (
    * Note that if there is a conflicting update on another node the entry will
    * not be removed after merge.
    */
+  def remove(node: SelfUniqueAddress, key: A): LWWMap[A, B] =
+    remove(node.uniqueAddress, key)
+
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def remove(node: Cluster, key: A): LWWMap[A, B] =
     remove(node.selfUniqueAddress, key)
 

akka-distributed-data/src/main/scala/akka/cluster/ddata/LWWRegister.scala

@@ -4,10 +4,10 @@
 
 package akka.cluster.ddata
 
+import akka.annotation.InternalApi
 import akka.cluster.Cluster
 import akka.cluster.UniqueAddress
 import akka.util.HashCode
-import akka.annotation.InternalApi
 
 object LWWRegister {
 
@@ -48,20 +48,48 @@ object LWWRegister {
   @InternalApi private[akka] def apply[A](node: UniqueAddress, initialValue: A, clock: Clock[A]): LWWRegister[A] =
     new LWWRegister(node, initialValue, clock(0L, initialValue))
 
+  def apply[A](node: SelfUniqueAddress, initialValue: A): LWWRegister[A] =
+    apply(node.uniqueAddress, initialValue, defaultClock[A])
+
+  def apply[A](node: SelfUniqueAddress, initialValue: A, clock: Clock[A]): LWWRegister[A] =
+    apply(node.uniqueAddress, initialValue, clock)
+
+  @deprecated("Use `apply` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def apply[A](initialValue: A)(implicit node: Cluster, clock: Clock[A] = defaultClock[A]): LWWRegister[A] =
     apply(node.selfUniqueAddress, initialValue, clock)
 
+  /**
+   * Scala API
+   * Creates a `LWWRegister` with implicits, given deprecated `apply` functions using Cluster constrain overloading.
+   */
+  def create[A](initialValue: A)(implicit node: SelfUniqueAddress, clock: Clock[A] = defaultClock[A]): LWWRegister[A] =
+    apply(node.uniqueAddress, initialValue, clock)
+
   /**
    * Java API
    */
+  @deprecated("Use `create` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def create[A](node: Cluster, initialValue: A): LWWRegister[A] =
     apply(initialValue)(node)
 
   /**
    * Java API
    */
+  @deprecated("Use `create` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def create[A](node: Cluster, initialValue: A, clock: Clock[A]): LWWRegister[A] =
-    apply(initialValue)(node, clock)
+    apply(node.selfUniqueAddress, initialValue, clock)
+
+  /**
+   * Java API
+   */
+  def create[A](node: SelfUniqueAddress, initialValue: A, clock: Clock[A]): LWWRegister[A] =
+    apply(node.uniqueAddress, initialValue, clock)
+
+  /**
+   * Java API
+   */
+  def create[A](node: SelfUniqueAddress, initialValue: A): LWWRegister[A] =
+    apply(node.uniqueAddress, initialValue, defaultClock[A])
 
   /**
    * Extract the [[LWWRegister#value]].
@@ -122,13 +150,13 @@ final class LWWRegister[A] private[akka] (
    * increasing version number from a database record that is used for optimistic
    * concurrency control.
    */
-  def withValue(value: A)(implicit node: Cluster, clock: Clock[A] = defaultClock[A]): LWWRegister[A] =
-    withValue(node, value, clock)
+  def withValue(node: SelfUniqueAddress, value: A, clock: Clock[A]): LWWRegister[A] =
+    withValue(node.uniqueAddress, value, clock)
 
   /**
    * Change the value of the register.
    */
-  def withValue(node: Cluster, value: A): LWWRegister[A] =
+  def withValue(node: SelfUniqueAddress, value: A): LWWRegister[A] =
     withValue(node, value, defaultClock[A])
 
   /**
@@ -139,6 +167,18 @@ final class LWWRegister[A] private[akka] (
    * increasing version number from a database record that is used for optimistic
    * concurrency control.
    */
+  def withValueOf(value: A)(implicit node: SelfUniqueAddress, clock: Clock[A] = defaultClock[A]): LWWRegister[A] =
+    withValue(node, value, clock)
+
+  @deprecated("Use `withValueOf` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def withValue(value: A)(implicit node: Cluster, clock: Clock[A] = defaultClock[A]): LWWRegister[A] =
+    withValue(node, value, clock)
+
+  @deprecated("Use `withValue` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def withValue(node: Cluster, value: A): LWWRegister[A] =
+    withValue(node, value, defaultClock[A])
+
+  @deprecated("Use `withValue` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def withValue(node: Cluster, value: A, clock: Clock[A]): LWWRegister[A] =
     withValue(node.selfUniqueAddress, value, clock)
 

akka-distributed-data/src/main/scala/akka/cluster/ddata/ORMap.scala

@@ -199,9 +199,15 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
    * Adds an entry to the map
    * @see [[#put]]
    */
+  def :+(entry: (A, B))(implicit node: SelfUniqueAddress): ORMap[A, B] = {
+    val (key, value) = entry
+    put(node.uniqueAddress, key, value)
+  }
+
+  @deprecated("Use `:+` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def +(entry: (A, B))(implicit node: Cluster): ORMap[A, B] = {
     val (key, value) = entry
-    put(node, key, value)
+    put(node.selfUniqueAddress, key, value)
   }
 
   /**
@@ -217,6 +223,9 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
    * value, because important history can be lost when replacing the `ORSet` and
    * undesired effects of merging will occur. Use [[ORMultiMap]] or [[#updated]] instead.
    */
+  def put(node: SelfUniqueAddress, key: A, value: B): ORMap[A, B] = put(node.uniqueAddress, key, value)
+
+  @deprecated("Use `put` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def put(node: Cluster, key: A, value: B): ORMap[A, B] = put(node.selfUniqueAddress, key, value)
 
   /**
@@ -241,6 +250,10 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
    * If there is no current value for the `key` the `initial` value will be
    * passed to the `modify` function.
    */
+  def updated(node: SelfUniqueAddress, key: A, initial: B)(modify: B ⇒ B): ORMap[A, B] =
+    updated(node.uniqueAddress, key, initial)(modify)
+
+  @deprecated("Use `updated` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def updated(node: Cluster, key: A, initial: B)(modify: B ⇒ B): ORMap[A, B] =
     updated(node.selfUniqueAddress, key, initial)(modify)
 
@@ -251,9 +264,9 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
    * passed to the `modify` function.
    */
   @Deprecated
-  @deprecated("use update for the Java API as updated is ambiguous with the Scala API", "2.5.19")
+  @deprecated("use update for the Java API as updated is ambiguous with the Scala API", "2.5.20")
   def updated(node: Cluster, key: A, initial: B, modify: java.util.function.Function[B, B]): ORMap[A, B] =
-    updated(node, key, initial)(value ⇒ modify.apply(value))
+    updated(node.selfUniqueAddress, key, initial)(value ⇒ modify.apply(value))
 
   /**
    * Java API: Replace a value by applying the `modify` function on the existing value.
@@ -261,6 +274,11 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
    * If there is no current value for the `key` the `initial` value will be
    * passed to the `modify` function.
    */
+  def update(node: SelfUniqueAddress, key: A, initial: B, modify: java.util.function.Function[B, B]): ORMap[A, B] =
+    updated(node.uniqueAddress, key, initial)(value ⇒ modify.apply(value))
+
+  @Deprecated
+  @deprecated("Use `update` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def update(node: Cluster, key: A, initial: B, modify: java.util.function.Function[B, B]): ORMap[A, B] =
     updated(node, key, initial)(value ⇒ modify.apply(value))
 
@@ -295,17 +313,25 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
   }
 
   /**
+   * Scala API
    * Removes an entry from the map.
    * Note that if there is a conflicting update on another node the entry will
    * not be removed after merge.
    */
-  def -(key: A)(implicit node: Cluster): ORMap[A, B] = remove(node, key)
+  def remove(key: A)(implicit node: SelfUniqueAddress): ORMap[A, B] = remove(node.uniqueAddress, key)
 
   /**
+   * Java API
    * Removes an entry from the map.
    * Note that if there is a conflicting update on another node the entry will
    * not be removed after merge.
    */
+  def remove(node: SelfUniqueAddress, key: A): ORMap[A, B] = remove(node.uniqueAddress, key)
+
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def -(key: A)(implicit node: Cluster): ORMap[A, B] = remove(node.selfUniqueAddress, key)
+
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def remove(node: Cluster, key: A): ORMap[A, B] = remove(node.selfUniqueAddress, key)
 
   /**

akka-distributed-data/src/main/scala/akka/cluster/ddata/ORMultiMap.scala

@@ -125,18 +125,28 @@ final class ORMultiMap[A, B] private[akka] (
   def size: Int = underlying.keys.elements.size
 
   /**
-   * Convenience for put. Requires an implicit Cluster.
+   * Convenience for put. Requires an implicit SelfUniqueAddress.
    * @see [[#put]]
    */
+  def :+(entry: (A, Set[B]))(implicit node: SelfUniqueAddress): ORMultiMap[A, B] = {
+    val (key, value) = entry
+    put(node.uniqueAddress, key, value)
+  }
+
+  @deprecated("Use `:+` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def +(entry: (A, Set[B]))(implicit node: Cluster): ORMultiMap[A, B] = {
     val (key, value) = entry
-    put(node, key, value)
+    put(node.selfUniqueAddress, key, value)
   }
 
   /**
    * Scala API: Associate an entire set with the key while retaining the history of the previous
    * replicated data set.
    */
+  def put(node: SelfUniqueAddress, key: A, value: Set[B]): ORMultiMap[A, B] =
+    put(node.uniqueAddress, key, value)
+
+  @deprecated("Use `put` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def put(node: Cluster, key: A, value: Set[B]): ORMultiMap[A, B] =
     put(node.selfUniqueAddress, key, value)
 
@@ -144,9 +154,16 @@ final class ORMultiMap[A, B] private[akka] (
    * Java API: Associate an entire set with the key while retaining the history of the previous
    * replicated data set.
    */
+  def put(node: SelfUniqueAddress, key: A, value: java.util.Set[B]): ORMultiMap[A, B] = {
+    import scala.collection.JavaConverters._
+    put(node.uniqueAddress, key, value.asScala.toSet)
+  }
+
+  @Deprecated
+  @deprecated("Use `put` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def put(node: Cluster, key: A, value: java.util.Set[B]): ORMultiMap[A, B] = {
     import scala.collection.JavaConverters._
-    put(node, key, value.asScala.toSet)
+    put(node.selfUniqueAddress, key, value.asScala.toSet)
   }
 
   /**
@@ -159,18 +176,27 @@ final class ORMultiMap[A, B] private[akka] (
     new ORMultiMap(newUnderlying, withValueDeltas)
   }
 
+  /**
+   * Scala API
+   * Remove an entire set associated with the key.
+   */
+  def remove(key: A)(implicit node: SelfUniqueAddress): ORMultiMap[A, B] = remove(node.uniqueAddress, key)
+
   /**
    * Convenience for remove. Requires an implicit Cluster.
    * @see [[#remove]]
    */
-  def -(key: A)(implicit node: Cluster): ORMultiMap[A, B] =
-    remove(node, key)
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def -(key: A)(implicit node: Cluster): ORMultiMap[A, B] = remove(node.selfUniqueAddress, key)
 
   /**
+   * Java API
    * Remove an entire set associated with the key.
    */
-  def remove(node: Cluster, key: A): ORMultiMap[A, B] =
-    remove(node.selfUniqueAddress, key)
+  def remove(node: SelfUniqueAddress, key: A): ORMultiMap[A, B] = remove(node.uniqueAddress, key)
+
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def remove(node: Cluster, key: A): ORMultiMap[A, B] = remove(node.selfUniqueAddress, key)
 
   /**
    * INTERNAL API
@@ -185,16 +211,22 @@ final class ORMultiMap[A, B] private[akka] (
   }
 
   /**
-   * Scala API: Add an element to a set associated with a key. If there is no existing set then one will be initialised.
+   * Add an element to a set associated with a key. If there is no existing set then one will be initialised.
+   * TODO add implicit after deprecated is EOL.
    */
+  def addBinding(node: SelfUniqueAddress, key: A, element: B): ORMultiMap[A, B] =
+    addBinding(node.uniqueAddress, key, element)
+
+  def addBindingBy(key: A, element: B)(implicit node: SelfUniqueAddress): ORMultiMap[A, B] =
+    addBinding(node, key, element)
+
+  @deprecated("Use `addBinding` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def addBinding(key: A, element: B)(implicit node: Cluster): ORMultiMap[A, B] =
     addBinding(node.selfUniqueAddress, key, element)
 
-  /**
-   * Java API: Add an element to a set associated with a key. If there is no existing set then one will be initialised.
-   */
+  @deprecated("Use `addBinding` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def addBinding(node: Cluster, key: A, element: B): ORMultiMap[A, B] =
-    addBinding(key, element)(node)
+    addBinding(node.selfUniqueAddress, key, element)
 
   /**
    * INTERNAL API
@@ -205,18 +237,24 @@ final class ORMultiMap[A, B] private[akka] (
   }
 
   /**
-   * Scala API: Remove an element of a set associated with a key. If there are no more elements in the set then the
+   * Remove an element of a set associated with a key. If there are no more elements in the set then the
    * entire set will be removed.
+   * TODO add implicit after deprecated is EOL.
    */
+  def removeBinding(node: SelfUniqueAddress, key: A, element: B): ORMultiMap[A, B] =
+    removeBinding(node.uniqueAddress, key, element)
+
+  def removeBindingBy(key: A, element: B)(implicit node: SelfUniqueAddress): ORMultiMap[A, B] =
+    removeBinding(node, key, element)
+
+  @deprecated("Use `removeBinding` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def removeBinding(key: A, element: B)(implicit node: Cluster): ORMultiMap[A, B] =
     removeBinding(node.selfUniqueAddress, key, element)
 
-  /**
-   * Java API: Remove an element of a set associated with a key. If there are no more elements in the set then the
-   * entire set will be removed.
-   */
+  @Deprecated
+  @deprecated("Use `removeBinding` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def removeBinding(node: Cluster, key: A, element: B): ORMultiMap[A, B] =
-    removeBinding(key, element)(node)
+    removeBinding(node.selfUniqueAddress, key, element)
 
   /**
    * INTERNAL API
@@ -241,6 +279,13 @@ final class ORMultiMap[A, B] private[akka] (
    * and another one is added within the same Update. The order of addition and removal is important in order
    * to retain history for replicated data.
    */
+  def replaceBinding(node: SelfUniqueAddress, key: A, oldElement: B, newElement: B): ORMultiMap[A, B] =
+    replaceBinding(node.uniqueAddress, key, oldElement, newElement)
+
+  def replaceBindingBy(key: A, oldElement: B, newElement: B)(implicit node: SelfUniqueAddress): ORMultiMap[A, B] =
+    replaceBinding(node, key, oldElement, newElement)
+
+  @deprecated("Use `replaceBinding` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def replaceBinding(key: A, oldElement: B, newElement: B)(implicit node: Cluster): ORMultiMap[A, B] =
     replaceBinding(node.selfUniqueAddress, key, oldElement, newElement)
 

akka-distributed-data/src/main/scala/akka/cluster/ddata/ORSet.scala

@@ -307,14 +307,17 @@ final class ORSet[A] private[akka] (
 
   def size: Int = elementsMap.size
 
-  /**
-   * Adds an element to the set
-   */
-  def +(element: A)(implicit node: Cluster): ORSet[A] = add(node, element)
+  /** Adds an element to the set. */
+  def :+(element: A)(implicit node: SelfUniqueAddress): ORSet[A] = add(node, element)
 
-  /**
-   * Adds an element to the set
-   */
+  @deprecated("Use `:+` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def +(element: A)(implicit node: Cluster): ORSet[A] = add(node.selfUniqueAddress, element)
+
+  /** Adds an element to the set. */
+  def add(node: SelfUniqueAddress, element: A): ORSet[A] = add(node.uniqueAddress, element)
+
+  @Deprecated
+  @deprecated("Use `add` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def add(node: Cluster, element: A): ORSet[A] = add(node.selfUniqueAddress, element)
 
   /**
@@ -335,13 +338,27 @@ final class ORSet[A] private[akka] (
   }
 
   /**
+   * Scala API
    * Removes an element from the set.
    */
-  def -(element: A)(implicit node: Cluster): ORSet[A] = remove(node, element)
+  def remove(element: A)(implicit node: SelfUniqueAddress): ORSet[A] = remove(node.uniqueAddress, element)
+
+  /**
+   * Java API
+   * Removes an element from the set.
+   */
+  def remove(node: SelfUniqueAddress, element: A): ORSet[A] = remove(node.uniqueAddress, element)
 
   /**
    * Removes an element from the set.
    */
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def -(element: A)(implicit node: Cluster): ORSet[A] = remove(node.selfUniqueAddress, element)
+
+  /**
+   * Removes an element from the set.
+   */
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def remove(node: Cluster, element: A): ORSet[A] = remove(node.selfUniqueAddress, element)
 
   /**
@@ -362,6 +379,9 @@ final class ORSet[A] private[akka] (
    * This has the same result as using [[#remove]] for each
    * element, but it is more efficient.
    */
+  def clear(node: SelfUniqueAddress): ORSet[A] = clear(node.uniqueAddress)
+
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def clear(node: Cluster): ORSet[A] = clear(node.selfUniqueAddress)
 
   /**

akka-distributed-data/src/main/scala/akka/cluster/ddata/PNCounter.scala

@@ -62,67 +62,98 @@ final class PNCounter private[akka] (
    * Increment the counter with the delta `n` specified.
    * If the delta is negative then it will decrement instead of increment.
    */
-  def +(n: Long)(implicit node: Cluster): PNCounter = increment(node, n)
+  def :+(n: Long)(implicit node: SelfUniqueAddress): PNCounter = increment(node.uniqueAddress, n)
+
+  @deprecated("Use `:+` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def +(n: Long)(implicit node: Cluster): PNCounter = increment(node.selfUniqueAddress, n)
 
   /**
    * Increment the counter with the delta `n` specified.
    * If the delta is negative then it will decrement instead of increment.
    */
-  def +(n: BigInt)(implicit node: Cluster): PNCounter = increment(node, n)
+  def :+(n: BigInt)(implicit node: SelfUniqueAddress): PNCounter = increment(node.uniqueAddress, n)
 
-  /**
-   * Increment the counter with the delta `n` specified.
-   * If the delta is negative then it will decrement instead of increment.
-   */
-  def increment(node: Cluster, n: Long = 1): PNCounter =
-    increment(node.selfUniqueAddress, n)
+  @deprecated("Use `:+` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def +(n: BigInt)(implicit node: Cluster): PNCounter = increment(node.selfUniqueAddress, n)
 
   /**
    * Scala API: Increment the counter with the delta `n` specified.
    * If the delta is negative then it will decrement instead of increment.
    */
-  def increment(node: Cluster, n: BigInt): PNCounter =
-    increment(node.selfUniqueAddress, n)
+  def increment(n: Long)(implicit node: SelfUniqueAddress): PNCounter = increment(node.uniqueAddress, n)
+
+  @deprecated("Use `increment` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def increment(node: Cluster, n: Long = 1): PNCounter = increment(node.selfUniqueAddress, n)
+
+  /**
+   * Increment the counter with the delta `n` specified.
+   * If the delta is negative then it will decrement instead of increment.
+   */
+  def increment(n: BigInt)(implicit node: SelfUniqueAddress): PNCounter = increment(node.uniqueAddress, n)
+
+  @deprecated("Use `increment` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def increment(node: Cluster, n: BigInt): PNCounter = increment(node.selfUniqueAddress, n)
 
   /**
    * Java API: Increment the counter with the delta `n` specified.
    * If the delta is negative then it will decrement instead of increment.
    */
-  def increment(node: Cluster, n: java.math.BigInteger): PNCounter =
-    increment(node.selfUniqueAddress, n)
+  def increment(node: SelfUniqueAddress, n: java.math.BigInteger): PNCounter = increment(node.uniqueAddress, n)
+
+  /**
+   * Java API: Increment the counter with the delta `n` specified.
+   * If the delta is negative then it will decrement instead of increment.
+   */
+  def increment(node: SelfUniqueAddress, n: Long): PNCounter = increment(node.uniqueAddress, n)
+
+  @deprecated("Use `increment` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def increment(node: Cluster, n: java.math.BigInteger): PNCounter = increment(node.selfUniqueAddress, n)
 
   /**
    * Decrement the counter with the delta `n` specified.
    * If the delta is negative then it will increment instead of decrement.
    */
-  def -(n: Long)(implicit node: Cluster): PNCounter = decrement(node, n)
+  def decrement(n: Long)(implicit node: SelfUniqueAddress): PNCounter = decrement(node.uniqueAddress, n)
+
+  @deprecated("Use `decrement` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def -(n: Long)(implicit node: Cluster): PNCounter = decrement(node.selfUniqueAddress, n)
 
   /**
    * Decrement the counter with the delta `n` specified.
    * If the delta is negative then it will increment instead of decrement.
    */
-  def -(n: BigInt)(implicit node: Cluster): PNCounter = decrement(node, n)
+  def decrement(n: BigInt)(implicit node: SelfUniqueAddress): PNCounter = decrement(node.uniqueAddress, n)
+
+  @deprecated("Use `decrement` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def -(n: BigInt)(implicit node: Cluster): PNCounter = decrement(node.selfUniqueAddress, n)
 
   /**
    * Decrement the counter with the delta `n` specified.
    * If the delta `n` is negative then it will increment instead of decrement.
    */
-  def decrement(node: Cluster, n: Long = 1): PNCounter =
-    decrement(node.selfUniqueAddress, n)
+  def decrement(node: SelfUniqueAddress, n: Long): PNCounter = decrement(node.uniqueAddress, n)
+
+  @deprecated("Use `decrement` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def decrement(node: Cluster, n: Long = 1): PNCounter = decrement(node.selfUniqueAddress, n)
 
   /**
    * Scala API: Decrement the counter with the delta `n` specified.
    * If the delta `n` is negative then it will increment instead of decrement.
    */
-  def decrement(node: Cluster, n: BigInt): PNCounter =
-    decrement(node.selfUniqueAddress, n)
+  def decrement(node: SelfUniqueAddress, n: BigInt): PNCounter = decrement(node.uniqueAddress, n)
+
+  @deprecated("Use `decrement` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def decrement(node: Cluster, n: BigInt): PNCounter = decrement(node.selfUniqueAddress, n)
 
   /**
    * Java API: Decrement the counter with the delta `n` specified.
    * If the delta `n` is negative then it will increment instead of decrement.
    */
-  def decrement(node: Cluster, n: java.math.BigInteger): PNCounter =
-    decrement(node.selfUniqueAddress, n)
+  def decrement(node: SelfUniqueAddress, n: java.math.BigInteger): PNCounter = decrement(node.uniqueAddress, n)
+
+  @Deprecated
+  @deprecated("Use `decrement` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def decrement(node: Cluster, n: java.math.BigInteger): PNCounter = decrement(node.selfUniqueAddress, n)
 
   /** Internal API */
   @InternalApi private[akka] def increment(key: UniqueAddress, n: BigInt): PNCounter = change(key, n)

akka-distributed-data/src/main/scala/akka/cluster/ddata/PNCounterMap.scala

@@ -4,11 +4,11 @@
 
 package akka.cluster.ddata
 
-import akka.cluster.Cluster
-import akka.cluster.UniqueAddress
 import java.math.BigInteger
 
 import akka.annotation.InternalApi
+import akka.cluster.Cluster
+import akka.cluster.UniqueAddress
 import akka.cluster.ddata.ORMap._
 
 object PNCounterMap {
@@ -75,13 +75,24 @@ final class PNCounterMap[A] private[akka] (
    * Increment the counter with the delta specified.
    * If the delta is negative then it will decrement instead of increment.
    */
-  def increment(key: A, delta: Long = 1)(implicit node: Cluster): PNCounterMap[A] =
-    increment(node, key, delta)
+  def incrementBy(key: A, delta: Long)(implicit node: SelfUniqueAddress): PNCounterMap[A] =
+    increment(node.uniqueAddress, key, delta)
 
   /**
    * Increment the counter with the delta specified.
    * If the delta is negative then it will decrement instead of increment.
    */
+  def increment(key: A, delta: Long = 1)(implicit node: Cluster): PNCounterMap[A] =
+    increment(node.selfUniqueAddress, key, delta)
+
+  /**
+   * Increment the counter with the delta specified.
+   * If the delta is negative then it will decrement instead of increment.
+   */
+  def increment(node: SelfUniqueAddress, key: A, delta: Long): PNCounterMap[A] =
+    increment(node.uniqueAddress, key, delta)
+
+  @deprecated("Use `increment` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def increment(node: Cluster, key: A, delta: Long): PNCounterMap[A] =
     increment(node.selfUniqueAddress, key, delta)
 
@@ -94,14 +105,28 @@ final class PNCounterMap[A] private[akka] (
   /**
    * Decrement the counter with the delta specified.
    * If the delta is negative then it will increment instead of decrement.
+   * TODO add implicit after deprecated is EOL.
    */
-  def decrement(key: A, delta: Long = 1)(implicit node: Cluster): PNCounterMap[A] =
+  def decrementBy(key: A, delta: Long = 1)(implicit node: SelfUniqueAddress): PNCounterMap[A] =
     decrement(node, key, delta)
 
   /**
    * Decrement the counter with the delta specified.
    * If the delta is negative then it will increment instead of decrement.
+   * TODO add implicit after deprecated is EOL.
    */
+  def decrement(node: SelfUniqueAddress, key: A, delta: Long): PNCounterMap[A] =
+    decrement(node.uniqueAddress, key, delta)
+
+  @deprecated("Use `decrement` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def decrement(key: A, delta: Long = 1)(implicit node: Cluster): PNCounterMap[A] =
+    decrement(node.selfUniqueAddress, key, delta)
+
+  /**
+   * Decrement the counter with the delta specified.
+   * If the delta is negative then it will increment instead of decrement.
+   */
+  @deprecated("Use `decrement` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def decrement(node: Cluster, key: A, delta: Long): PNCounterMap[A] =
     decrement(node.selfUniqueAddress, key, delta)
 
@@ -117,16 +142,16 @@ final class PNCounterMap[A] private[akka] (
    * Note that if there is a conflicting update on another node the entry will
    * not be removed after merge.
    */
-  def -(key: A)(implicit node: Cluster): PNCounterMap[A] = remove(node, key)
+  def remove(key: A)(implicit node: SelfUniqueAddress): PNCounterMap[A] =
+    remove(node.uniqueAddress, key)
 
-  /**
-   * Removes an entry from the map.
-   * Note that if there is a conflicting update on another node the entry will
-   * not be removed after merge.
-   */
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def remove(node: Cluster, key: A): PNCounterMap[A] =
     remove(node.selfUniqueAddress, key)
 
+  @deprecated("Use `remove` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def -(key: A)(implicit node: Cluster): PNCounterMap[A] = remove(node, key)
+
   /**
    * INTERNAL API
    */

akka-distributed-data/src/main/scala/akka/cluster/ddata/Replicator.scala

@@ -100,6 +100,15 @@ object ReplicatorSettings {
    */
   @InternalApi private[akka] def roleOption(role: String): Option[String] =
     if (role == "") None else Option(role)
+
+  /**
+   * INTERNAL API
+   * The name of the actor used in DistributedData extensions.
+   */
+  @InternalApi private[akka] def name(system: ActorSystem, modifier: Option[String]): String = {
+    val name = system.settings.config.getString("akka.cluster.distributed-data.name")
+    modifier.map(s ⇒ s + name.take(1).toUpperCase + name.drop(1)).getOrElse(name)
+  }
 }
 
 /**

akka-distributed-data/src/main/scala/akka/cluster/ddata/VersionVector.scala

@@ -9,7 +9,6 @@ import scala.annotation.tailrec
 import scala.collection.immutable.TreeMap
 import akka.cluster.Cluster
 import akka.cluster.UniqueAddress
-import akka.cluster.UniqueAddress
 import akka.annotation.InternalApi
 
 /**
@@ -107,7 +106,10 @@ sealed abstract class VersionVector
   /**
    * Increment the version for the node passed as argument. Returns a new VersionVector.
    */
-  def +(node: Cluster): VersionVector = increment(node)
+  def :+(node: SelfUniqueAddress): VersionVector = increment(node)
+
+  @deprecated("Use `:+` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
+  def +(node: Cluster): VersionVector = increment(node.selfUniqueAddress)
 
   /**
    * INTERNAL API
@@ -118,6 +120,9 @@ sealed abstract class VersionVector
   /**
    * Increment the version for the node passed as argument. Returns a new VersionVector.
    */
+  def increment(node: SelfUniqueAddress): VersionVector = increment(node.uniqueAddress)
+
+  @deprecated("Use `increment` that takes a `SelfUniqueAddress` parameter instead.", since = "2.5.20")
   def increment(node: Cluster): VersionVector = increment(node.selfUniqueAddress)
 
   def isEmpty: Boolean