pjfanning commented on code in PR #3677:
URL: https://github.com/apache/pekko/pull/3677#discussion_r4366421092
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala:
##########
@@ -233,13 +243,14 @@ object ShardCoordinator {
* @param shardId the id of the shard to allocate
* @param currentShardAllocations all actor refs to `ShardRegion` and
their current allocated shards,
* in the order they were allocated
- * @return a `Future` of the actor ref of the `ShardRegion` that is to be
responsible for the shard, must be one of
- * the references included in the `currentShardAllocations`
parameter
+ * @return a `CompletionStage` of the actor ref of the `ShardRegion` that
is to be responsible for the shard,
+ * must be one of the references included in the
`currentShardAllocations` parameter
+ * (in Pekko 1.x, this was a Scala `Future`)
*/
def allocateShard(
requester: ActorRef,
shardId: String,
- currentShardAllocations: java.util.Map[ActorRef,
immutable.IndexedSeq[String]]): Future[ActorRef]
+ currentShardAllocations: java.util.Map[ActorRef, ? <:
java.lang.Iterable[String]]): CompletionStage[ActorRef]
Review Comment:
`java.util.Map<ActorRef, ? extends Iterable<String>>` gives Java
implementers no `size()`, `get(i)` or `isEmpty()`, which typical strategies
(least-shards, a port of `LeastShardAllocationStrategy`) need. Since
`asJavaShardAllocations` already produces `java.util.List` values, could this
be `java.util.Map<ActorRef, java.util.List<String>>`? That also drops the
wildcard, which is awkward to work with on the Java side. Same for `rebalance`.
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala:
##########
@@ -212,17 +214,25 @@ object ShardCoordinator {
requester: ActorRef,
shardId: ShardId,
currentShardAllocations: Map[ActorRef,
immutable.IndexedSeq[ShardId]]): Future[ActorRef] = {
-
- import scala.jdk.CollectionConverters._
- allocateShard(requester, shardId, currentShardAllocations.asJava)
+ import scala.jdk.FutureConverters._
+ allocateShard(requester, shardId,
asJavaShardAllocations(currentShardAllocations)).asScala
}
override final def rebalance(
currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardId]],
rebalanceInProgress: Set[ShardId]): Future[Set[ShardId]] = {
import scala.jdk.CollectionConverters._
+ import scala.jdk.FutureConverters._
implicit val ec = ExecutionContext.parasitic
- rebalance(currentShardAllocations.asJava,
rebalanceInProgress.asJava).map(_.asScala.toSet)
+ rebalance(asJavaShardAllocations(currentShardAllocations),
rebalanceInProgress.asJava)
+ .asScala
+ .map(_.asScala.toSet)
+ }
+
+ private def asJavaShardAllocations(currentShardAllocations: Map[ActorRef,
immutable.IndexedSeq[ShardId]])
+ : java.util.Map[ActorRef, java.util.List[ShardId]] = {
+ import scala.jdk.CollectionConverters._
+ currentShardAllocations.map { case (region, shards) => region ->
shards.asJava }.asJava
}
Review Comment:
This eagerly builds a new map (O(regions)) on every
`allocateShard`/`rebalance` call, where the previous `.asJava` was an O(1)
wrapper. During mass allocation in a large cluster that's extra garbage on the
coordinator's single-threaded path. A lazy view (e.g.
`currentShardAllocations.view.mapValues(_.asJava)` wrapped as a
`java.util.Map`, or a small `AbstractMap` that wraps values on access) would
avoid the copy.
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala:
##########
@@ -233,13 +243,14 @@ object ShardCoordinator {
* @param shardId the id of the shard to allocate
* @param currentShardAllocations all actor refs to `ShardRegion` and
their current allocated shards,
* in the order they were allocated
- * @return a `Future` of the actor ref of the `ShardRegion` that is to be
responsible for the shard, must be one of
- * the references included in the `currentShardAllocations`
parameter
+ * @return a `CompletionStage` of the actor ref of the `ShardRegion` that
is to be responsible for the shard,
+ * must be one of the references included in the
`currentShardAllocations` parameter
+ * (in Pekko 1.x, this was a Scala `Future`)
*/
def allocateShard(
Review Comment:
A Java subclass compiled against 1.x won't implement the new abstract
`CompletionStage` overloads, so the final Scala `allocateShard`/`rebalance`
will throw `AbstractMethodError` inside the coordinator and allocation stops.
The `ReversedMissingMethodProblem` excludes hide this from MiMa. That may be
acceptable for 2.0, but if so the migration guide should say explicitly that
compiled strategies must be recompiled. The alternative is a new abstract class
alongside a deprecated old one.
##########
docs/src/main/paradox/migration/migration-guide-1.x-2.x.md:
##########
@@ -65,6 +65,13 @@ by Akka 2.3.x. The `-2.3` program argument and the
`remove2dot3Data` parameter o
`Sequence`. Code that cast these offsets to `Sequence` needs updating; the
per-persistence-id sequence number is
still available as `envelope.sequenceNr`. These queries also now accept a
`TimestampOffset` to resume from.
([PR3615](https://github.com/apache/pekko/pull/3615))
+* `ShardCoordinator.AbstractShardAllocationStrategy`, the Java API for custom
shard allocation strategies, no longer
+leaks Scala types. Its `allocateShard` and `rebalance` methods now return a
`java.util.concurrent.CompletionStage`
+instead of a Scala `Future`, and the current shard allocations are passed as a
+`java.util.Map<ActorRef, ? extends Iterable<String>>` instead of a
`java.util.Map` of
+`scala.collection.immutable.IndexedSeq<String>`. Java implementations of these
two methods need to be updated
+accordingly; Scala implementations of
`ShardCoordinator.ShardAllocationStrategy` are unaffected.
Review Comment:
Suggest stating that Java strategies compiled against Pekko 1.x must be
recompiled: otherwise they fail at runtime with `AbstractMethodError` when the
coordinator calls them, rather than at compile/link time. (And if the parameter
type changes to `java.util.List<String>`, update this text.)
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala:
##########
@@ -212,17 +214,25 @@ object ShardCoordinator {
requester: ActorRef,
shardId: ShardId,
currentShardAllocations: Map[ActorRef,
immutable.IndexedSeq[ShardId]]): Future[ActorRef] = {
-
- import scala.jdk.CollectionConverters._
- allocateShard(requester, shardId, currentShardAllocations.asJava)
+ import scala.jdk.FutureConverters._
+ allocateShard(requester, shardId,
asJavaShardAllocations(currentShardAllocations)).asScala
}
override final def rebalance(
currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardId]],
rebalanceInProgress: Set[ShardId]): Future[Set[ShardId]] = {
import scala.jdk.CollectionConverters._
+ import scala.jdk.FutureConverters._
implicit val ec = ExecutionContext.parasitic
- rebalance(currentShardAllocations.asJava,
rebalanceInProgress.asJava).map(_.asScala.toSet)
+ rebalance(asJavaShardAllocations(currentShardAllocations),
rebalanceInProgress.asJava)
+ .asScala
+ .map(_.asScala.toSet)
+ }
+
+ private def asJavaShardAllocations(currentShardAllocations: Map[ActorRef,
immutable.IndexedSeq[ShardId]])
+ : java.util.Map[ActorRef, java.util.List[ShardId]] = {
+ import scala.jdk.CollectionConverters._
+ currentShardAllocations.map { case (region, shards) => region ->
shards.asJava }.asJava
Review Comment:
Nit: `CollectionConverters`/`FutureConverters` are imported locally in three
places within ~20 lines; a single import at the top of the class would be
tidier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
