Implementing Akka Cluster Sharding for Stateful Persistent Entities (Scala)

Desired outcome

Deploy a resilient Akka Cluster where stateful entities are distributed across nodes via Cluster Sharding, tolerate node restarts, and recover shard locations automatically.

Prerequisites

1. Akka 2.8.x (typed) core, cluster, and sharding libraries – all artifacts must share the same version.
2. A running Akka Cluster (seed nodes or akka‑management bootstrap).
3. A persistent journal and snapshot store configured (e.g., Akka Persistence JDBC or Cassandra).
4. Build tool (sbt or Maven) and JDK 11+.
5. Network access between nodes on the Akka remoting port (default 25520) and management port (default 8558).

Focused procedure

1. Add dependencies

In build.sbt (Scala) or pom.xml (Java) include:

libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-actor-typed" % "2.8.0",
  "com.typesafe.akka" %% "akka-cluster-typed" % "2.8.0",
  "com.typesafe.akka" %% "akka-cluster-sharding-typed" % "2.8.0",
  "com.typesafe.akka" %% "akka-persistence-typed" % "2.8.0",
  "com.typesafe.akka" %% "akka-persistence-jdbc" % "5.0.0"   // example journal
)

2. Configure the cluster and sharding

Create src/main/resources/application.conf:

akka {
  actor {
    provider = cluster
  }
  remote {
    artery {
      canonical.host = "${AKKA_HOST}"   # set via env or config
      canonical.port = 25520
    }
  }
  cluster {
    seed-nodes = ["akka://[contact removed]:25520", "akka://[contact removed]:25520"]
    # or use akka-management bootstrap
  }
  cluster.sharding {
    # use ddata for lightweight coordinator state; change to persistence if needed
    coordinating-distributed-data = on
    # optional: tune rebalance interval
    handoff-timeout = 60s
  }
  persistence {
    journal.plugin = "akka.persistence.jdbc.journal"
    snapshot-store.plugin = "akka.persistence.jdbc.snapshot-store"
    jdbc {
      # journal & snapshot config omitted for brevity
    }
  }
}

3. Define the entity behavior

Example of a persistent, typed entity:

object CounterEntity {
  sealed trait Command
  final case class Increment(replyTo: ActorRef[Done]) extends Command
  final case class GetValue(replyTo: ActorRef[Int]) extends Command
  private case class State(value: Int) extends CborSerializable

  def apply(entityId: String): Behavior[Command] = {
    EventSourcedBehavior[Command, Event, State](
      persistenceId = PersistenceId.of("CounterEntity", entityId),
      emptyState = State(0),
      commandHandler = (state, cmd) => cmd match {
        case Increment(replyTo) =>
          Effect.persist(Incremented).thenRun(_ => replyTo ! Done)
        case GetValue(replyTo) => Effect.reply(replyTo)(state.value)
      },
      eventHandler = (state, evt) => {
        evt match {
          case Incremented => State(state.value + 1)
        }
      }
    )
  }

  private sealed trait Event extends CborSerializable
  private case object Incremented extends Event
}

4. Initialize sharding on each node

In your system bootstrap (e.g., Main.scala):

val system = ActorSystem(Behaviors.empty, "my-system")
val sharding = ClusterSharding(system)

val typeKey = EntityTypeKey[CounterEntity.Command]("counter")

sharding.init(Entity(typeKey)(createBehavior = entityCtx => CounterEntity(entityCtx.entityId)))

// Example usage: send a message to entity "user-42"
val entityId = "user-42"
sharding.entityRefFor(typeKey, entityId) ! CounterEntity.Increment(replyTo)

5. Enable passivation (optional but recommended)

To avoid indefinite memory growth, configure an idle timeout:

val settings = ClusterShardingSettings(system)
  .withPassivationStrategySettings(
    PassivationStrategySettings(idleEntityStrategy = IdleEntityStrategy.afterTimeout(5.minutes))
  )

sharding.init(Entity(typeKey)(createBehavior = entityCtx => CounterEntity(entityCtx.entityId)).withSettings(settings))

Expected checks

• Entity activation: On startup, logs should show Starting entity counter|user-42 on the node that owns the shard.
• Message delivery: After sending Increment, the entity logs the persisted Incremented event and replies Done.
• Shard rebalancing: When a node leaves the cluster, the ShardCoordinator logs Shard [counter|10] migrating from nodeA to nodeB. Messages buffered during handoff are redelivered after the entity restarts on the new node.
• Passivation verification: With a 5‑minute idle timeout, after no messages the entity logs Passivating counter|user-42 and later Starting entity counter|user-42 on next message, restoring state from the journal.

Recovery options

Coordinator state loss

If the ddata or persistence store used by the ShardCoordinator is corrupted, shards will be reallocated from scratch on the next cluster join. Entities will recover their state from the journal because persistenceId includes the entity ID.

Stuck passivation

Ensure the entity respects the Passivate signal (Akka typed does this automatically). If you implement a custom stop message, verify that the entity replies with the stop acknowledgment; otherwise increase the handoff timeout in application.conf.

Diagnosing rebalance stalls

Enable debug logging for akka.cluster.sharding and watch for ShardHomeAllocated and ShardHomeAllocationFailed messages. The akka‑management HTTP endpoint (GET /cluster/members and GET /cluster/sharding) shows current shard allocations.

Limitations and practical verification

• Sharding provides at‑least‑once delivery; entity code must be idempotent or detect duplicates (e.g., by tracking processed event IDs).
• All Akka modules (core, cluster, sharding, persistence) must match exactly; mixing versions leads to NoSuchMethodError or silent misrouting.
• The message extractor (implicit in the typed API) must be deterministic; changing the entityId extraction logic after deployment breaks routing to existing entities.
• Changing the coordinator mode (ddata ↔ persistence) requires a full cluster restart; it cannot be altered on a live cluster.

To verify the setup locally:

1. Start two seed nodes with different AKKA_HOST values (e.g., 127.0.0.1 and 127.0.0.2) using sbt run.
2. From a third terminal, run a simple client that sends Increment to several entity IDs (e.g., user-1 through user-20).
3. Check each node’s logs for entity activation lines; roughly half the entities should appear on each node.
4. Stop one node (Ctrl+C) and observe the ShardCoordinator rebalancing logs; then resend messages to the same IDs and confirm they are processed on the remaining node.
5. Enable passivation, wait >5 minutes without traffic, then send a new message; the log should show a fresh activation with state restored from the journal.

Rollback (if needed)

Since initializing sharding creates persistent journal entries, a clean rollback involves:

1. Stopping all nodes.
2. Removing the journal and snapshot tables (or dropping the keyspace) for the persistence plugin.
3. Clearing the ddata directory if using ddata coordinator.
4. Restarting the cluster with a clean slate.

Only perform these steps in a test environment; in production, prefer to scale down gracefully rather than delete persisted state.