Skip to content

chore: improvements around dynamic scaling of sharded daemon processes #1437

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
johanandren wants to merge 2 commits into
base: main
Choose a base branch
from
Open

chore: improvements around dynamic scaling of sharded daemon processes #1437

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
fe146e6
chore: improvements around dynamic scaling of sharded daemon processes
johanandren May 20, 2026
565d085
review feedback addressed
johanandren May 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions ...ction-grpc/src/main/scala/akka/projection/grpc/replication/internal/ReplicationImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,6 @@ private[akka] object ReplicationImpl {
projectionName.size < 255,
s"The generated projection name for replica [${remoteReplica.replicaId.id}]: '$projectionName' is too long to fit " +
"in the database column, must be at most 255 characters. See if you can shorten replica or entity type names.")
val sliceRanges = Persistence(system).sliceRanges(remoteReplica.numberOfConsumers)

val grpcQuerySettings = {
val s = GrpcQuerySettings(settings.streamId).withFromReplica(remoteReplica.replicaId)
Expand Down Expand Up @@ -207,9 +206,11 @@ private[akka] object ReplicationImpl {
case Some(role) => defaultWithShardingSettings.withRole(role)
}
}
ShardedDaemonProcess(system).init(sanitizeActorName(projectionName), remoteReplica.numberOfConsumers, {
idx =>
val sliceRange = sliceRanges(idx)
ShardedDaemonProcess(system).initWithContext[ProjectionBehavior.Command](
sanitizeActorName(projectionName),
remoteReplica.numberOfConsumers, { context =>
val sliceRanges = Persistence(system).sliceRanges(context.totalProcesses)
val sliceRange = sliceRanges(context.processNumber)
val projectionKey = s"${sliceRange.min}-${sliceRange.max}"
val projectionId = ProjectionId(projectionName, projectionKey)

Expand Down Expand Up @@ -319,7 +320,9 @@ private[akka] object ReplicationImpl {
sliceRange.min,
sliceRange.max)
ProjectionBehavior(settings.projectionProvider(projectionId, sourceProvider, replicationFlow, system))
}, shardedDaemonProcessSettings, Some(ProjectionBehavior.Stop))
},
shardedDaemonProcessSettings,
ProjectionBehavior.Stop)
}

/**
Expand Down Expand Up @@ -368,7 +371,6 @@ private[akka] object ReplicationImpl {
projectionName.size < 255,
s"The generated projection name for replication: '$projectionName' is too long to fit " +
"in the database column, must be at most 255 characters. See if you can shorten replica or entity type names.")
val sliceRanges = Persistence(system).sliceRanges(remoteReplica.numberOfConsumers)

val shardedDaemonProcessSettings = {
import scala.concurrent.duration._
Expand Down Expand Up @@ -415,6 +417,7 @@ private[akka] object ReplicationImpl {
sanitizeActorName(s"${settings.selfReplicaId.id}EventProducer"),
// FIXME separate setting for number of producers?
remoteReplica.numberOfConsumers, { (context: ShardedDaemonProcessContext) =>
val sliceRanges = Persistence(system).sliceRanges(context.totalProcesses)
val sliceRange = sliceRanges(context.processNumber)
val projectionKey = s"${sliceRange.min}-${sliceRange.max}"
val projectionId = ProjectionId(projectionName, projectionKey)
Expand Down
4 changes: 3 additions & 1 deletion docs/src/main/paradox/grpc.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,9 @@ The gRPC connection to the producer is defined in the [consumer configuration](#
The @ref:[R2dbcProjection](r2dbc.md) has support for storing the offset in a relational database using R2DBC.

The above example is using the @extref:[ShardedDaemonProcess](akka:typed/cluster-sharded-daemon-process.html) to distribute the instances of the Projection across the cluster.
There are alternative ways of running the `ProjectionBehavior` as described in @ref:[Running a Projection](running.md)
There are alternative ways of running the `ProjectionBehavior` as described in @ref:[Running a Projection](running.md).

Note that the `numberOfProjectionInstances` value is only honored the first time the Sharded Daemon Process is started; afterwards the running count is kept in distributed data. See @ref:[Changing the number of projection instances](running.md#changing-the-number-of-projection-instances) for details.

How to implement the `EventHandler` and choose between different processing semantics is described in the @ref:[R2dbcProjection documentation](r2dbc.md).

Expand Down
10 changes: 10 additions & 0 deletions docs/src/main/paradox/running.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,16 @@ For this example, we configure as many `ShardedDaemonProcess` as tags and we def

For graceful stop it is recommended to use @scala[`ProjectionBehavior.Stop`]@java[`ProjectionBehavior.stop()`] message.

### Changing the number of projection instances

@@@ warning
The `numberOfInstances`/`initialNumberOfInstances` argument passed to `ShardedDaemonProcess.init`/`initWithContext` is only used the first time a given Sharded Daemon Process is started. After that the running count is kept in distributed data and is authoritative — redeploying with a different argument value alone will **not** change the number of running projection instances.

To change the number at runtime send the `ChangeNumberOfProcesses` command to the `ActorRef` returned from `initWithContext`, as described in @extref:[Sharded Daemon Process dynamic scaling](akka:typed/cluster-sharded-daemon-process.html#dynamic-scaling-of-number-of-workers).

When the behavior factory derives slice ranges from the configured number (the common `sliceRanges(numberOfConsumers)` pattern), compute the slice ranges *inside* the behavior factory from `daemonContext.totalProcesses` rather than from the outer `numberOfInstances` value. Otherwise a stale distributed data state (from a previous deployment that rescaled to a different number) will cause processes outside the configured range to start with an out-of-bounds slice index and fail in a restart loop.
@@@

### Projection Behavior

The `ProjectionBehavior` is an Actor `Behavior` that knows how to manage the Projection lifecyle. The Projection starts to consume the events as soon as the actor is spawned and will restart the source in case of failures (see @ref:[Projection Settings](projection-settings.md)).
Expand Down
20 changes: 12 additions & 8 deletions samples/grpc/iot-service-java/src/main/java/iot/temperature/TemperatureEvents.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import akka.Done;
import akka.actor.typed.ActorSystem;
import akka.cluster.sharding.typed.ShardedDaemonProcessSettings;
import akka.cluster.sharding.typed.javadsl.ClusterSharding;
import akka.cluster.sharding.typed.javadsl.ShardedDaemonProcess;
import akka.http.javadsl.model.HttpRequest;
Expand Down Expand Up @@ -81,17 +82,20 @@ public static void initPushedEventsConsumer(ActorSystem<?> system) {
var numberOfSliceRanges =
system.settings().config().getInt("iot-service.temperature.projections-slice-count");

var sliceRanges =
EventSourcedProvider.sliceRanges(
system, R2dbcReadJournal.Identifier(), numberOfSliceRanges);

ShardedDaemonProcess.get(system)
.init(
.initWithContext(

@patriknw patriknw May 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we want to promote initWithContext in all places, even if scaling isn't used/needed? the signature looks more complex than init?

@johanandren johanandren May 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, good point, I went for all the places to make sure anything users could copy paste is safe to evolve/scale, but maybe that is overdoing it.

ProjectionBehavior.Command.class,
"TemperatureProjection",
sliceRanges.size(),
i -> ProjectionBehavior.create(projection(system, sliceRanges.get(i))),
ProjectionBehavior.stopMessage());
numberOfSliceRanges,
daemonContext -> {
var sliceRanges =
EventSourcedProvider.sliceRanges(
system, R2dbcReadJournal.Identifier(), daemonContext.totalProcesses());

@patriknw patriknw May 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we use EventSourcedProvider here and in some other places

Persistence(system).sliceRanges(context.totalProcesses)

would be easier to always use Persistence.sliceRanges?

@johanandren johanandren May 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree, strange

@johanandren johanandren May 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

21 places across projections samples and specs (!!)

@johanandren johanandren May 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What it adds is looking up the read journal for that specific plugin/config, so that the actual journal could have some other slicing scheme.

But in practice it is always the same, so should we always use the simpler one? In that case, why does the EventSourcedProvider#sliceRanges even exist?

PR adding the method didn't shed any light on it #609

@patriknw patriknw May 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, I would guess that the guy thought it would be nice to stay within projections api surface and not reach out to persistence. It's typically used together with

        EventSourcedProvider.eventsBySlices(
            system,
            R2dbcReadJournal.Identifier(),
            PRODUCER_ENTITY_TYPE,
            sliceRange.first(),
            sliceRange.second());

Let's leave it as is, using EventSourcedProvider#sliceRanges

@patriknw patriknw May 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might look better to move the whole slice range stuff into the projection method, and just pass the totalProcesses to that method, but not very important.

return ProjectionBehavior.create(
projection(system, sliceRanges.get(daemonContext.processNumber())));
},
ShardedDaemonProcessSettings.create(system),
Optional.of(ProjectionBehavior.stopMessage()));
}

private static Projection<EventEnvelope<TemperatureRead>> projection(
Expand Down
6 changes: 2 additions & 4 deletions samples/grpc/iot-service-scala/src/main/scala/iot/registration/Registration.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,8 +61,7 @@ object Registration {

sealed trait Event extends CborSerializable

final case class Registered(secret: SecretDataValue)
extends Event
final case class Registered(secret: SecretDataValue) extends Event

val EntityKey: EntityTypeKey[Command] =
EntityTypeKey[Command]("Registration")
Expand All @@ -77,8 +76,7 @@ object Registration {
.withEnforcedReplies[Command, Event, State](
persistenceId = PersistenceId(EntityKey.name, entityId),
emptyState = State.empty,
commandHandler =
(state, command) => handleCommand(state, command),
commandHandler = (state, command) => handleCommand(state, command),
eventHandler = (state, event) => handleEvent(state, event))
.withRetention(RetentionCriteria.snapshotEvery(numberOfEvents = 100))
.onPersistFailure(
Expand Down
3 changes: 2 additions & 1 deletion samples/grpc/iot-service-scala/src/main/scala/iot/registration/RegistrationServiceImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@ class RegistrationServiceImpl(system: ActorSystem[_])

override def register(in: proto.RegisterRequest): Future[Empty] = {
logger.info("register sensor {}", in.sensorEntityId)
val entityRef = sharding.entityRefFor(Registration.EntityKey, in.sensorEntityId)
val entityRef =
sharding.entityRefFor(Registration.EntityKey, in.sensorEntityId)
val reply: Future[Done] =
entityRef.askWithStatus(
Registration.Register(Registration.SecretDataValue(in.secret), _))
Expand Down
3 changes: 2 additions & 1 deletion samples/grpc/iot-service-scala/src/main/scala/iot/temperature/SensorTwinServiceImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,8 @@ class SensorTwinServiceImpl(system: ActorSystem[_]) extends SensorTwinService {

override def getTemperature(
in: proto.GetTemperatureRequest): Future[proto.CurrentTemperature] = {
val entityRef = sharding.entityRefFor(SensorTwin.EntityKey, in.sensorEntityId)
val entityRef =
sharding.entityRefFor(SensorTwin.EntityKey, in.sensorEntityId)
val reply: Future[Int] =
entityRef.askWithStatus(SensorTwin.GetTemperature(_))
val response =
Expand Down
21 changes: 12 additions & 9 deletions samples/grpc/iot-service-scala/src/main/scala/iot/temperature/TemperatureEvents.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package iot.temperature
import scala.concurrent.Future

import akka.actor.typed.ActorSystem
import akka.cluster.sharding.typed.ShardedDaemonProcessSettings
import akka.cluster.sharding.typed.scaladsl.ClusterSharding
import akka.cluster.sharding.typed.scaladsl.ShardedDaemonProcess
import akka.http.scaladsl.model.HttpRequest
Expand Down Expand Up @@ -39,8 +40,7 @@ object TemperatureEvents {
TemperatureEventsStreamId,
proto.TemperatureEventsProto.javaDescriptor.getFile :: Nil)

EventProducerPushDestination
.grpcServiceHandler(destination)(system)
EventProducerPushDestination.grpcServiceHandler(destination)(system)
}

def initPushedEventsConsumer(implicit system: ActorSystem[_]): Unit = {
Expand Down Expand Up @@ -98,15 +98,18 @@ object TemperatureEvents {
// Split the slices into N ranges
val numberOfSliceRanges: Int = system.settings.config
.getInt("iot-service.temperature.projections-slice-count")
val sliceRanges = EventSourcedProvider.sliceRanges(
system,
R2dbcReadJournal.Identifier,
numberOfSliceRanges)

ShardedDaemonProcess(system).init(
ShardedDaemonProcess(system).initWithContext(
name = "TemperatureProjection",
numberOfInstances = sliceRanges.size,
behaviorFactory = i => ProjectionBehavior(projection(sliceRanges(i))),
initialNumberOfInstances = numberOfSliceRanges,
behaviorFactory = { daemonContext =>
val sliceRanges = EventSourcedProvider.sliceRanges(
system,
R2dbcReadJournal.Identifier,
daemonContext.totalProcesses)
ProjectionBehavior(projection(sliceRanges(daemonContext.processNumber)))
},
settings = ShardedDaemonProcessSettings(system),
stopMessage = ProjectionBehavior.Stop)

}
Expand Down
4 changes: 2 additions & 2 deletions samples/grpc/iot-service-scala/src/test/scala/iot/EdgeApp.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,8 +110,8 @@ object EdgeApp {
SensorSimulator.TemperatureRead,
temperature.proto.TemperatureRead] { envelope =>
val event = envelope.event
Future.successful(Some(
temperature.proto.TemperatureRead(event.temperature)))
Future.successful(
Some(temperature.proto.TemperatureRead(event.temperature)))
}

val eventProducer = EventProducerPush[SensorSimulator.Event](
Expand Down
15 changes: 8 additions & 7 deletions samples/grpc/local-drone-control-java/src/main/java/local/drones/DroneEvents.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
import akka.projection.grpc.producer.javadsl.EventProducerSource;
import akka.projection.grpc.producer.javadsl.Transformation;
import akka.projection.r2dbc.javadsl.R2dbcProjection;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import org.slf4j.Logger;
Expand Down Expand Up @@ -84,7 +83,6 @@ public static Behavior<ProjectionBehavior.Command> eventToCloudPushBehavior(
public static void initEventToCloudDaemonProcess(ActorSystem<Void> system, Settings settings) {
var nrOfEventProducers =
system.settings().config().getInt("local-drone-control.nr-of-event-producers");
var sliceRanges = Persistence.get(system).getSliceRanges(nrOfEventProducers);

// turn events into a public protocol (protobuf) type before publishing
var eventTransformation =
Expand Down Expand Up @@ -114,19 +112,22 @@ public static void initEventToCloudDaemonProcess(ActorSystem<Void> system, Setti
GrpcClientSettings.fromConfig("central-drone-control", system));

ShardedDaemonProcess.get(system)
.init(
.initWithContext(
ProjectionBehavior.Command.class,
"drone-event-push",
nrOfEventProducers,
idx -> projectionForPartition(system, eventProducer, sliceRanges, idx));
daemonContext -> {
var sliceRanges =
Persistence.get(system).getSliceRanges(daemonContext.totalProcesses());
return projectionForPartition(
system, eventProducer, sliceRanges.get(daemonContext.processNumber()));
});
}

private static Behavior<ProjectionBehavior.Command> projectionForPartition(
ActorSystem<?> system,
EventProducerPush<Object> eventProducer,
List<Pair<Integer, Integer>> sliceRanges,
int partition) {
var sliceRange = sliceRanges.get(partition);
Pair<Integer, Integer> sliceRange) {
var minSlice = sliceRange.first();
var maxSlice = sliceRange.second();

Expand Down
55 changes: 26 additions & 29 deletions samples/grpc/local-drone-control-scala/src/main/scala/local/drones/DroneEvents.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,6 @@ object DroneEvents {

val nrOfEventProducers =
system.settings.config.getInt("local-drone-control.nr-of-event-producers")
val sliceRanges = Persistence(system).sliceRanges(nrOfEventProducers)

// turn events into a public protocol (protobuf) type before publishing
val eventTransformation =
Expand All @@ -112,36 +111,34 @@ object DroneEvents {
envelope.event.isInstanceOf[Drone.CoarseGrainedLocationChanged]),
GrpcClientSettings.fromConfig("central-drone-control"))

def projectionForPartition(
partition: Int): Behavior[ProjectionBehavior.Command] = {
val sliceRange = sliceRanges(partition)
val minSlice = sliceRange.min
val maxSlice = sliceRange.max

ProjectionBehavior(
R2dbcProjection.atLeastOnceFlow[Offset, EventEnvelope[Drone.Event]](
ProjectionId("drone-event-push", s"$minSlice-$maxSlice"),
settings = None,
sourceProvider = EventSourcedProvider
.eventsBySlicesStartingFromSnapshots[Drone.State, Drone.Event](
system,
R2dbcReadJournal.Identifier,
eventProducer.eventProducerSource.entityType,
minSlice,
maxSlice,
// start from latest drone snapshot and don't replay history
{ (state: Drone.State) =>
Drone.CoarseGrainedLocationChanged(
state.coarseGrainedCoordinates.get)
}),
handler = eventProducer.handler()))

}

ShardedDaemonProcess(system).init(
ShardedDaemonProcess(system).initWithContext(
"drone-event-push",
nrOfEventProducers,
projectionForPartition)
{ daemonContext =>
val sliceRanges =
Persistence(system).sliceRanges(daemonContext.totalProcesses)
val sliceRange = sliceRanges(daemonContext.processNumber)
val minSlice = sliceRange.min
val maxSlice = sliceRange.max

ProjectionBehavior(
R2dbcProjection.atLeastOnceFlow[Offset, EventEnvelope[Drone.Event]](
ProjectionId("drone-event-push", s"$minSlice-$maxSlice"),
settings = None,
sourceProvider = EventSourcedProvider
.eventsBySlicesStartingFromSnapshots[Drone.State, Drone.Event](
system,
R2dbcReadJournal.Identifier,
eventProducer.eventProducerSource.entityType,
minSlice,
maxSlice,
// start from latest drone snapshot and don't replay history
{ (state: Drone.State) =>
Drone.CoarseGrainedLocationChanged(
state.coarseGrainedCoordinates.get)
}),
handler = eventProducer.handler()))
})

}

Expand Down
20 changes: 12 additions & 8 deletions ...staurant-drone-deliveries-service-java/src/main/java/central/drones/LocalDroneEvents.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import akka.Done;
import akka.actor.typed.ActorSystem;
import akka.cluster.sharding.typed.ShardedDaemonProcessSettings;
import akka.cluster.sharding.typed.javadsl.ClusterSharding;
import akka.cluster.sharding.typed.javadsl.ShardedDaemonProcess;
import akka.japi.Pair;
Expand Down Expand Up @@ -110,17 +111,20 @@ public static void initPushedEventsConsumer(ActorSystem<?> system) {
.config()
.getInt("restaurant-drone-deliveries-service.drones.projections-slice-count");

var sliceRanges =
EventSourcedProvider.sliceRanges(
system, R2dbcReadJournal.Identifier(), numberOfSliceRanges);

ShardedDaemonProcess.get(system)
.init(
.initWithContext(
ProjectionBehavior.Command.class,
"LocalDronesProjection",
sliceRanges.size(),
i -> ProjectionBehavior.create(projection(system, sliceRanges.get(i))),
ProjectionBehavior.stopMessage());
numberOfSliceRanges,
daemonContext -> {
var sliceRanges =
EventSourcedProvider.sliceRanges(
system, R2dbcReadJournal.Identifier(), daemonContext.totalProcesses());
return ProjectionBehavior.create(
projection(system, sliceRanges.get(daemonContext.processNumber())));
},
ShardedDaemonProcessSettings.create(system),
Optional.of(ProjectionBehavior.stopMessage()));
}

private static Projection<EventEnvelope<CoarseDroneLocation>> projection(
Expand Down
17 changes: 10 additions & 7 deletions ...one-deliveries-service-scala/src/main/scala/central/deliveries/RestaurantDeliveries.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,13 +82,16 @@ object RestaurantDeliveries {
PersistenceId(EntityKey.name, restaurantId),
None,
onCommand,
onEvent).withTaggerForState {
case (Some(state), _) =>
// tag events with location id as topic, grpc projection filters makes sure only that location
// picks them up for drone delivery
Set("t:" + state.localControlLocationId)
case _ => Set.empty
}.onPersistFailure(SupervisorStrategy.restartWithBackoff(100.millis, 5.seconds, 0.1))
onEvent)
.withTaggerForState {
case (Some(state), _) =>
// tag events with location id as topic, grpc projection filters makes sure only that location
// picks them up for drone delivery
Set("t:" + state.localControlLocationId)
case _ => Set.empty
}
.onPersistFailure(
SupervisorStrategy.restartWithBackoff(100.millis, 5.seconds, 0.1))

// #commandHandler
private def onCommand(
Expand Down
Loading