You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2026/10/24 01:00:51 UTC
[PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
He-Pin opened a new pull request, #3677:
URL: https://github.com/apache/pekko/pull/3677
### Motivation
`ShardCoordinator.AbstractShardAllocationStrategy` is the Java API for custom shard allocation strategies, but its two abstract methods returned a Scala `Future` and took the allocated shards as `scala.collection.immutable.IndexedSeq`. Java implementers had to work with Scala types.
Pekko 2.0.0 removes such leaks from the Java APIs elsewhere (for example pekko-persistence in #2092), and a scan of `cluster-sharding` and `cluster-sharding-typed` shows this was the last Java API in the sharding modules that still leaked a Scala `Future` or a Scala collection.
### Modification
* `allocateShard` and `rebalance` now return `java.util.concurrent.CompletionStage` instead of `scala.concurrent.Future`.
* The current shard allocations are passed as a `java.util.Map` of `java.lang.Iterable` shard ids instead of `scala.collection.immutable.IndexedSeq`. `java.lang.Iterable` follows the existing precedent for input parameters in `pekko.dispatch.CompletionStages` and the router constructors. `rebalanceInProgress` stays a `java.util.Set` because its contract is membership testing, and the `rebalance` result stays a `java.util.Set` because it is an output.
* The `final` Scala-facing overrides convert between the Scala `ShardAllocationStrategy` API and the Java API using `scala.jdk.FutureConverters`/`CollectionConverters`. The Scala `ShardAllocationStrategy` trait itself is unchanged.
* Added MiMa filters for the two changed abstract methods.
Note that deprecating the `Future`-returning methods instead of replacing them is not possible here: the old and new `allocateShard`/`rebalance` overloads would differ only in their return type, which `javac` rejects for methods with the same erasure. This PR therefore follows the approach already taken for the pekko-persistence Java APIs in #2092 and documents the change in the migration guide.
### Result
A custom shard allocation strategy can now be written in Java using Java types only. Scala implementations of `ShardAllocationStrategy` are unaffected. Existing Java subclasses have to switch their return types and shard-id collections; this is recorded in the 1.x to 2.x migration guide.
### Tests
* `sbt "cluster-sharding/Test/testOnly org.apache.pekko.cluster.sharding.AbstractShardAllocationStrategySpec"`: 5 tests passed. The new spec drives a Java implementation (`RecordingShardAllocationStrategy`) through the Scala API and asserts the forwarded requester, shard id and allocation order, the `rebalanceInProgress` contents, the converted results, and that a failed `CompletionStage` surfaces as a failed `Future`.
* `sbt "cluster-sharding/mimaReportBinaryIssues"`: passed with the new filters (Scala 2.13).
* `sbt headerCreateAll`, `sbt "cluster-sharding/Test/javafmt"`: passed.
* Native `scalafmt` on the changed Scala files, and `git diff --check`: passed.
* `docs/paradox` for the migration guide and release notes change.
* Not run locally: JDK 17 is not installed, so the Java formatter ran on JDK 21. The Scala 3 `mimaReportBinaryIssues` run and `sbt validatePullRequest` are delegated to CI.
### References
Refs #2092 - continues removing Scala types from the Pekko Java APIs for 2.0.0
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #3677:
URL: https://github.com/apache/pekko/pull/3677#issuecomment-6385123768
Both halves are in now. `RecordingShardAllocationStrategy` implements `StartableAllocationStrategy`, the integration test asserts `start()` is called exactly once and before any `allocateShard` or `rebalance`, and the `AbstractShardAllocationStrategy` Javadoc points Java implementers at the trait and notes that `start()` must not block. The migration guide mentions it too.
I did not add a separate `start()` hook on the abstract class. Implementing the existing Scala trait from Java works - it compiles to a single-method interface, so a Java class just declares `implements StartableAllocationStrategy` and `public void start()` - and a second hook would duplicate it while making the coordinator call `start()` on every Java strategy whether it wants the callback or not.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #3677:
URL: https://github.com/apache/pekko/pull/3677#discussion_r4366422477
##########
cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/AbstractShardAllocationStrategySpec.scala:
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.cluster.sharding
+
+import scala.collection.immutable
+import scala.jdk.CollectionConverters._
+
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.cluster.sharding.ShardRegion.ShardId
+import pekko.testkit.PekkoSpec
+import pekko.testkit.TestProbe
+
+class AbstractShardAllocationStrategySpec extends PekkoSpec {
+
+ private val regionA: ActorRef = TestProbe().ref
+ private val regionB: ActorRef = TestProbe().ref
+
+ private val allocations: Map[ActorRef, immutable.IndexedSeq[ShardId]] = Map(
+ regionA -> immutable.IndexedSeq("shard-1", "shard-2"),
+ regionB -> immutable.IndexedSeq("shard-3"))
+
+ private val expectedAllocations: Map[ActorRef, List[ShardId]] =
+ allocations.map { case (region, shards) => region -> shards.toList }
+
+ private def seenAllocations(strategy: RecordingShardAllocationStrategy): Map[ActorRef, List[ShardId]] = {
+ val seen = strategy.allocationsSeen.asScala
+ seen should have size 1
+ seen.head.asScala.iterator.map { case (region, shards) => region -> shards.asScala.toList }.toMap
+ }
+
+ "The Java API of AbstractShardAllocationStrategy" must {
Review Comment:
These tests only call the adapter methods directly, so nothing runs a Java `AbstractShardAllocationStrategy` inside a real `ShardCoordinator`. A small spec that starts sharding with `RecordingShardAllocationStrategy` via `ClusterSharding.start(..., allocationStrategy, ...)` and checks that the strategy is invoked and shards get allocated would catch integration regressions, for example a later change to how the coordinator calls or wraps the strategy.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3677:
URL: https://github.com/apache/pekko/pull/3677#discussion_r4365440094
##########
cluster-sharding/src/main/scala/org/apache/pekko/cluster/sharding/ShardCoordinator.scala:
##########
@@ -233,13 +245,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:
The PR description promises migration guidance, but `docs/src/main/paradox/migration/migration-guide-1.x-2.x.md` has no entry for this breaking Java API change. Add instructions for updating both methods to return `CompletionStage`, changing allocation values from `IndexedSeq<String>` to `Iterable<String>`, and recompiling existing subclasses. Include the release-note update mentioned in the description.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #3677:
URL: https://github.com/apache/pekko/pull/3677#discussion_r4366440393
##########
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:
This is an input parameter, and you can pass the List in too. Not the return one
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on PR #3677:
URL: https://github.com/apache/pekko/pull/3677#issuecomment-6381584503
Since this PR aims to make `AbstractShardAllocationStrategy` a complete Java API ahead of the 2.0 freeze: there is currently no documented or tested way for a Java strategy to also hook into `StartableAllocationStrategy.start()`. A Java implementer would have to implement the Scala trait on top of `AbstractShardAllocationStrategy`. That probably works, but it is undocumented and has no test. Could you either add a test (e.g. have `RecordingShardAllocationStrategy` implement `StartableAllocationStrategy` and assert `start()` is called) and mention it in the Javadoc, or expose a Java-friendly `start()` hook on the abstract class?
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #3677:
URL: https://github.com/apache/pekko/pull/3677#discussion_r4367348684
##########
cluster-sharding/src/test/scala/org/apache/pekko/cluster/sharding/AbstractShardAllocationStrategySpec.scala:
##########
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.cluster.sharding
+
+import scala.collection.immutable
+import scala.jdk.CollectionConverters._
+
+import org.apache.pekko
+import pekko.actor.ActorRef
+import pekko.cluster.sharding.ShardRegion.ShardId
+import pekko.testkit.PekkoSpec
+import pekko.testkit.TestProbe
+
+class AbstractShardAllocationStrategySpec extends PekkoSpec {
+
+ private val regionA: ActorRef = TestProbe().ref
+ private val regionB: ActorRef = TestProbe().ref
+
+ private val allocations: Map[ActorRef, immutable.IndexedSeq[ShardId]] = Map(
+ regionA -> immutable.IndexedSeq("shard-1", "shard-2"),
+ regionB -> immutable.IndexedSeq("shard-3"))
+
+ private val expectedAllocations: Map[ActorRef, List[ShardId]] =
+ allocations.map { case (region, shards) => region -> shards.toList }
+
+ private def seenAllocations(strategy: RecordingShardAllocationStrategy): Map[ActorRef, List[ShardId]] = {
+ val seen = strategy.allocationsSeen.asScala
+ seen should have size 1
+ seen.head.asScala.iterator.map { case (region, shards) => region -> shards.asScala.toList }.toMap
+ }
+
+ "The Java API of AbstractShardAllocationStrategy" must {
Review Comment:
Added - `"start a Java strategy before allocating and rebalancing through the coordinator"` starts a real single-node cluster and sharding via `ClusterSharding.start(..., allocationStrategy = strategy, ...)`, sends two entities, and asserts that `start()` ran exactly once and first, that both shards were allocated through the Java callback, that a rebalance round reached the strategy, and that the coordinator's allocation view exposes the shard ids as a `java.util.List` of size 2.
The assertions all live inside one `awaitAssert` so they are checked against a single snapshot of the recorded calls.
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Use CompletionStage in the AbstractShardAllocationStrategy Java API [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #3677:
URL: https://github.com/apache/pekko/pull/3677
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org