Classic Cluster Singleton

For the full documentation of this feature and for new projects see Cluster Singleton.

Module info

To use Cluster Singleton, you must add the following dependency in your project:

sbt

val PekkoVersion = "1.1.2+29-e21fa9eb-SNAPSHOT"
libraryDependencies += "org.apache.pekko" %% "pekko-cluster-tools" % PekkoVersion

Maven

<properties>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.apache.pekko</groupId>
      <artifactId>pekko-bom_${scala.binary.version}</artifactId>
      <version>1.1.2+29-e21fa9eb-SNAPSHOT</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>
<dependencies>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-cluster-tools_${scala.binary.version}</artifactId>
  </dependency>
</dependencies>

Gradle

def versions = [
  ScalaBinary: "2.13"
]
dependencies {
  implementation platform("org.apache.pekko:pekko-bom_${versions.ScalaBinary}:1.1.2+29-e21fa9eb-SNAPSHOT")

  implementation "org.apache.pekko:pekko-cluster-tools_${versions.ScalaBinary}"
}

Project Info: Pekko Cluster Tools (classic)
Artifact	org.apache.pekko pekko-cluster-tools 1.1.2+29-e21fa9eb-SNAPSHOT Snapshots are available
JDK versions	OpenJDK 8 OpenJDK 11 OpenJDK 17 OpenJDK 21
Scala versions	2.13.14, 2.12.20, 3.3.4
JPMS module name	pekko.cluster.tools
License	Apache-2.0
Home page	https://pekko.apache.org/
API documentation	API (Scaladoc) API (Javadoc)
Forums	Apache Pekko User mailing list Apache Pekko Dev mailing list apache/pekko discussion
Release notes	Release Notes
Issues	Github issues
Sources	https://github.com/apache/pekko

Introduction

For the full documentation of this feature and for new projects see Cluster Singleton - Introduction.

The cluster singleton pattern is implemented by org.apache.pekko.cluster.singleton.ClusterSingletonManager. It manages one singleton actor instance among all cluster nodes or a group of nodes tagged with a specific role. ClusterSingletonManager is an actor that is supposed to be started as early as possible on all nodes, or all nodes with specified role, in the cluster. The actual singleton actor is started by the ClusterSingletonManager on the oldest node by creating a child actor from supplied Props. ClusterSingletonManager makes sure that at most one singleton instance is running at any point in time.

You can access the singleton actor by using the provided org.apache.pekko.cluster.singleton.ClusterSingletonProxy, which will route all messages to the current instance of the singleton. The proxy will keep track of the oldest node in the cluster and resolve the singleton’s ActorRef by explicitly sending the singleton’s actorSelection the org.apache.pekko.actor.Identify message and waiting for it to reply. This is performed periodically if the singleton doesn’t reply within a certain (configurable) time. Given the implementation, there might be periods of time during which the ActorRef is unavailable, e.g., when a node leaves the cluster. In these cases, the proxy will buffer the messages sent to the singleton and then deliver them when the singleton is finally available. If the buffer is full the ClusterSingletonProxy will drop old messages when new messages are sent via the proxy. The size of the buffer is configurable and it can be disabled by using a buffer size of 0.

See Cluster Singleton - Potential problems to be aware of.

An Example

Assume that we need one single entry point to an external system. An actor that receives messages from a JMS queue with the strict requirement that only one JMS consumer must exist to make sure that the messages are processed in order. That is perhaps not how one would like to design things, but a typical real-world scenario when integrating with external systems.

Before explaining how to create a cluster singleton actor, let’s define message classes and their corresponding factory methods which will be used by the singleton.

Scala

copysourceobject PointToPointChannel {
  case object UnregistrationOk extends CborSerializable
}
object Consumer {
  case object End extends CborSerializable
  case object GetCurrent extends CborSerializable
  case object Ping extends CborSerializable
  case object Pong extends CborSerializable
}

Java

copysourcepublic class TestSingletonMessages {
  public static class UnregistrationOk {}

  public static class End {}

  public static class Ping {}

  public static class Pong {}

  public static class GetCurrent {}

  public static UnregistrationOk unregistrationOk() {
    return new UnregistrationOk();
  }

  public static End end() {
    return new End();
  }

  public static Ping ping() {
    return new Ping();
  }

  public static Pong pong() {
    return new Pong();
  }

  public static GetCurrent getCurrent() {
    return new GetCurrent();
  }
}

On each node in the cluster you need to start the ClusterSingletonManager and supply the Props of the singleton actor, in this case the JMS queue consumer.

Scala

copysourcesystem.actorOf(
  ClusterSingletonManager.props(
    singletonProps = Props(classOf[Consumer], queue, testActor),
    terminationMessage = End,
    settings = ClusterSingletonManagerSettings(system).withRole("worker")),
  name = "consumer")

Java

copysourcefinal ClusterSingletonManagerSettings settings =
    ClusterSingletonManagerSettings.create(system).withRole("worker");

system.actorOf(
    ClusterSingletonManager.props(
        Props.create(Consumer.class, () -> new Consumer(queue, testActor)),
        TestSingletonMessages.end(),
        settings),
    "consumer");

Here we limit the singleton to nodes tagged with the "worker" role, but all nodes, independent of role, can be used by not specifying withRole.

We use an application specific terminationMessage (i.e. TestSingletonMessages.end() message) to be able to close the resources before actually stopping the singleton actor. Note that PoisonPill is a perfectly fine terminationMessage if you only need to stop the actor.

Here is how the singleton actor handles the terminationMessage in this example.

Scala

copysourcecase End =>
  queue ! UnregisterConsumer
case UnregistrationOk =>
  stoppedBeforeUnregistration = false
  context.stop(self)
case Ping =>
  sender() ! Pong

Java

copysource.match(End.class, message -> queue.tell(UnregisterConsumer.class, getSelf()))
.match(
    UnregistrationOk.class,
    message -> {
      stoppedBeforeUnregistration = false;
      getContext().stop(getSelf());
    })
.match(Ping.class, message -> getSender().tell(TestSingletonMessages.pong(), getSelf()))

With the names given above, access to the singleton can be obtained from any cluster node using a properly configured proxy.

Scala

copysourceval proxy = system.actorOf(
  ClusterSingletonProxy.props(
    singletonManagerPath = "/user/consumer",
    settings = ClusterSingletonProxySettings(system).withRole("worker")),
  name = "consumerProxy")

Java

copysourceClusterSingletonProxySettings proxySettings =
    ClusterSingletonProxySettings.create(system).withRole("worker");

ActorRef proxy =
    system.actorOf(
        ClusterSingletonProxy.props("/user/consumer", proxySettings), "consumerProxy");

Configuration

For the full documentation of this feature and for new projects see Cluster Singleton - configuration.

Supervision

There are two actors that could potentially be supervised. For the consumer singleton created above these would be:

• Cluster singleton manager e.g. /user/consumer which runs on every node in the cluster
• The user actor e.g. /user/consumer/singleton which the manager starts on the oldest node

The Cluster singleton manager actor should not have its supervision strategy changed as it should always be running. However, it is sometimes useful to add supervision for the user actor. To accomplish this, add a parent supervisor actor which will be used to create the ‘real’ singleton instance. Below is an example implementation (credit to this StackOverflow answer)

Scala

copysourceimport org.apache.pekko.actor.{ Actor, Props, SupervisorStrategy }
class SupervisorActor(childProps: Props, override val supervisorStrategy: SupervisorStrategy) extends Actor {
  val child = context.actorOf(childProps, "supervised-child")

  def receive = {
    case msg => child.forward(msg)
  }
}

Java

copysourceimport org.apache.pekko.actor.AbstractActor;
import org.apache.pekko.actor.AbstractActor.Receive;
import org.apache.pekko.actor.ActorRef;
import org.apache.pekko.actor.Props;
import org.apache.pekko.actor.SupervisorStrategy;

public class SupervisorActor extends AbstractActor {
  final Props childProps;
  final SupervisorStrategy supervisorStrategy;
  final ActorRef child;

  SupervisorActor(Props childProps, SupervisorStrategy supervisorStrategy) {
    this.childProps = childProps;
    this.supervisorStrategy = supervisorStrategy;
    this.child = getContext().actorOf(childProps, "supervised-child");
  }

  @Override
  public SupervisorStrategy supervisorStrategy() {
    return supervisorStrategy;
  }

  @Override
  public Receive createReceive() {
    return receiveBuilder().matchAny(msg -> child.forward(msg, getContext())).build();
  }
}

And used here

Scala

copysourceimport org.apache.pekko
import pekko.actor.{ PoisonPill, Props }
import pekko.cluster.singleton.{ ClusterSingletonManager, ClusterSingletonManagerSettings }
context.system.actorOf(
  ClusterSingletonManager.props(
    singletonProps = Props(classOf[SupervisorActor], props, supervisorStrategy),
    terminationMessage = PoisonPill,
    settings = ClusterSingletonManagerSettings(context.system)),
  name = name)

Java

copysourceimport org.apache.pekko.actor.PoisonPill;
import org.apache.pekko.actor.Props;
import org.apache.pekko.cluster.singleton.ClusterSingletonManager;
import org.apache.pekko.cluster.singleton.ClusterSingletonManagerSettings;

copysourcereturn getContext()
    .system()
    .actorOf(
        ClusterSingletonManager.props(
            Props.create(
                SupervisorActor.class, () -> new SupervisorActor(props, supervisorStrategy)),
            PoisonPill.getInstance(),
            ClusterSingletonManagerSettings.create(getContext().system())),
        name = name);

Lease

For the full documentation of this feature and for new projects see Cluster Singleton - Lease.