He-Pin commented on code in PR #3677:
URL: https://github.com/apache/pekko/pull/3677#discussion_r4367348199
##########
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:
Switched to `java.util.Map<ActorRef, ? extends java.util.List<String>>`.
You were right and my earlier reply answered a different question.
`java.lang.Iterable` has no `size()`, so a least-shards strategy would have to
walk every shard id of every region on each `allocateShard` - O(total shards)
on the coordinator's single-threaded path. That is the same cost the lazy view
was meant to remove, so the view only fixed half of it.
Bounding the wildcard on `List` rather than dropping it keeps both
properties:
* implementations reach the `List` API with no downcast -
`entry.getValue().size()`, `map.get(region).get(0)` and `isEmpty()` all
compile. `AbstractShardAllocationStrategyTest.exposeAllocationValuesAsLists`
asserts this;
* callers can still pass `Map<ActorRef, ArrayList<String>>` or `LinkedList`
- `acceptArrayListValues` / `acceptLinkedListValues`.
Both directions are verified with javac, not assumed. One consequence is now
in the Javadoc: a Java override cannot narrow the parameter to `Map<ActorRef,
List<String>>`, javac rejects it with "have the same erasure, yet neither
overrides the other", so overrides must keep the wildcard.
`asJavaShardAllocations` already produced `java.util.List` values, so the
adapter needed no change. Migration guide and the MiMa filter comment updated.
The filters themselves are unchanged and still sufficient - generic parameters
do not appear in the descriptor, so `Iterable` -> `List` adds no new problem;
`+cluster-sharding/mimaReportBinaryIssues` passes on 2.13.18 and 3.3.8 against
1.0.0.
##########
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:
Fixed - `asJavaShardAllocations` now returns a `java.util.AbstractMap` that
wraps values on access, the second of your two suggestions. `get`,
`containsKey` and `size` are O(1) and only the entries a strategy actually
iterates allocate a wrapper.
`AbstractShardAllocationStrategySpec` has a regression test for it: it hands
the adapter a map whose `iterator` throws, so any eager traversal fails the
suite.
##########
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:
Done - both converters are imported once at the top of
`AbstractShardAllocationStrategy` instead of three times inside the methods.
##########
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:
Documented rather than adding a parallel deprecated class. The migration
guide now states explicitly that a subclass compiled against Pekko 1.x does not
implement the new callbacks and fails at runtime with `AbstractMethodError`
when the coordinator invokes the strategy, preventing allocation and
rebalancing, and that Java subclasses must be recompiled against Pekko 2.x.
That matches how the other Java API cleanups for 2.0 were handled.
##########
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:
Done - the entry now says the failure is a runtime `AbstractMethodError`
when the coordinator calls the strategy, not a compile or link time error, and
the parameter type reads `java.util.Map<ActorRef, ? extends List<String>>`.
--
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]
