This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch port/1ce8748239-state-store-plugin-docs in repository https://gitbox.apache.org/repos/asf/pekko.git
commit d93d475d7b416dc1cfea3e889cb4e79d53048178 Author: He-Pin(kerr) <[email protected]> AuthorDate: Tue Jun 30 14:33:36 2026 +0800 docs: how to write a custom Durable State store plugin Motivation: Pekko Persistence supports pluggable Durable State storage backends, but there was no documentation explaining how to implement one. Plugin authors had to read the source to discover the `DurableStateUpdateStore` and `DurableStateStoreProvider` SPI. Modification: Add a "Building a storage backend for Durable State" paradox page with Scala and Java examples (plugin API, provider, and minimal activation configuration), backed by a runnable doc spec. Also fix a stale scaladoc on `DurableStateStoreProvider.scaladslDurableStateStore` that incorrectly referred to `ReadJournal` instead of `DurableStateStore`. Improvements over the upstream example: the Java `MyJavaStateStore.upsertObject` uses the generic type parameter `A` instead of `Object`, so the snippet is a correct, type-safe template when adapted to a concrete state type; and the provider uses the diamond operator to avoid raw-type usage. Deliberately divergent from upstream: Pekko's `DurableStateUpdateStore` already documents `@param revision` (better than the upstream `@param seqNr` fix), and the upstream `//#plugin-api` markers on the real trait are not referenced by the docs (the snippet is taken from the example), so they are omitted to avoid clutter in production sources. Result: Durable State plugin authors have a step-by-step guide with compiling, tested Scala and Java examples. Tests: - sbt "docs/testOnly docs.persistence.state.PersistenceStatePluginDocSpec" (2/2 passed) - sbt "docs/paradox" (success; all snippets resolve) References: Upstream: akka/akka-core@1ce8748239e968f1fba15d23213130e5be98945b (akka#31973), which is now Apache licensed. --- .../docs/persistence/state/MyJavaStateStore.java | 77 ++++++++++++++++++++ .../state/MyJavaStateStoreProvider.java | 45 ++++++++++++ .../paradox/durable-state/state-store-plugin.md | 45 ++++++++++++ .../typed/index-persistence-durable-state.md | 1 + .../docs/persistence/state/MyStateStore.scala | 68 ++++++++++++++++++ .../state/PersistenceStatePluginDocSpec.scala | 84 ++++++++++++++++++++++ .../state/DurableStateStoreProvider.scala | 2 +- 7 files changed, 321 insertions(+), 1 deletion(-) diff --git a/docs/src/main/java/docs/persistence/state/MyJavaStateStore.java b/docs/src/main/java/docs/persistence/state/MyJavaStateStore.java new file mode 100644 index 0000000000..b643557253 --- /dev/null +++ b/docs/src/main/java/docs/persistence/state/MyJavaStateStore.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * license agreements; and to You under the Apache License, version 2.0: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * This file is part of the Apache Pekko project, which was derived from Akka. + */ + +/* + * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com> + */ + +package docs.persistence.state; + +// #plugin-imports + +import org.apache.pekko.Done; +import org.apache.pekko.actor.ExtendedActorSystem; +import org.apache.pekko.persistence.state.javadsl.DurableStateUpdateStore; +import org.apache.pekko.persistence.state.javadsl.GetObjectResult; +import com.typesafe.config.Config; +import java.util.concurrent.CompletionStage; + +// #plugin-imports + +// #state-store-plugin-api +class MyJavaStateStore<A> implements DurableStateUpdateStore<A> { + + private ExtendedActorSystem system; + private Config config; + private String cfgPath; + + public MyJavaStateStore(ExtendedActorSystem system, Config config, String cfgPath) { + this.system = system; + this.config = config; + this.cfgPath = cfgPath; + } + + /** Returns the current state for the given persistence id. */ + @Override + public CompletionStage<GetObjectResult<A>> getObject(String persistenceId) { + // implement getObject here + return null; + } + + /** + * Will persist the latest state. If it's a new persistence id, the record will be inserted. + * + * <p>In case of an existing persistence id, the record will be updated only if the revision + * number of the incoming record is 1 more than the already existing record. Otherwise persist + * will fail. + */ + @Override + public CompletionStage<Done> upsertObject( + String persistenceId, long revision, A value, String tag) { + // implement upsertObject here + return null; + } + + /** Deprecated. Use the deleteObject overload with revision instead. */ + @Override + public CompletionStage<Done> deleteObject(String persistenceId) { + return deleteObject(persistenceId, 0); + } + + /** + * Will delete the state by setting it to the empty state and the revision number will be + * incremented by 1. + */ + @Override + public CompletionStage<Done> deleteObject(String persistenceId, long revision) { + // implement deleteObject here + return null; + } +} +// #state-store-plugin-api diff --git a/docs/src/main/java/docs/persistence/state/MyJavaStateStoreProvider.java b/docs/src/main/java/docs/persistence/state/MyJavaStateStoreProvider.java new file mode 100644 index 0000000000..cfa42fce7a --- /dev/null +++ b/docs/src/main/java/docs/persistence/state/MyJavaStateStoreProvider.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * license agreements; and to You under the Apache License, version 2.0: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * This file is part of the Apache Pekko project, which was derived from Akka. + */ + +/* + * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com> + */ + +package docs.persistence.state; + +import org.apache.pekko.actor.ExtendedActorSystem; +import org.apache.pekko.persistence.state.DurableStateStoreProvider; +import org.apache.pekko.persistence.state.scaladsl.DurableStateStore; +import com.typesafe.config.Config; + +// #plugin-provider +class MyJavaStateStoreProvider implements DurableStateStoreProvider { + + private ExtendedActorSystem system; + private Config config; + private String cfgPath; + + public MyJavaStateStoreProvider(ExtendedActorSystem system, Config config, String cfgPath) { + this.system = system; + this.config = config; + this.cfgPath = cfgPath; + } + + @Override + public DurableStateStore<Object> scaladslDurableStateStore() { + return new MyStateStore<>(this.system, this.config, this.cfgPath); + } + + @Override + public org.apache.pekko.persistence.state.javadsl.DurableStateStore<Object> + javadslDurableStateStore() { + return new MyJavaStateStore<>(this.system, this.config, this.cfgPath); + } +} +// #plugin-provider diff --git a/docs/src/main/paradox/durable-state/state-store-plugin.md b/docs/src/main/paradox/durable-state/state-store-plugin.md new file mode 100644 index 0000000000..584ef30846 --- /dev/null +++ b/docs/src/main/paradox/durable-state/state-store-plugin.md @@ -0,0 +1,45 @@ +# Building a storage backend for Durable State + +Storage backends for state stores are pluggable in the Pekko persistence extension. +This documentation describes how to build a new storage backend for durable state. + +Applications can provide their own plugins by implementing a plugin API and activating them by configuration. +Plugin development requires the following imports: + +Scala +: @@snip [PersistenceStatePluginDocSpec.scala](/docs/src/test/scala/docs/persistence/state/PersistenceStatePluginDocSpec.scala) { #plugin-imports } + +Java +: @@snip [MyJavaStateStore.java](/docs/src/main/java/docs/persistence/state/MyJavaStateStore.java) { #plugin-imports } + +## State Store plugin API + +A durable state store plugin extends `DurableStateUpdateStore`. + +`DurableStateUpdateStore` is an interface and the methods to be implemented are: + +Scala +: @@snip [MyStateStore.scala](/docs/src/main/scala/docs/persistence/state/MyStateStore.scala) { #plugin-api } + +Java +: @@snip [MyJavaStateStore.java](/docs/src/main/java/docs/persistence/state/MyJavaStateStore.java) { #state-store-plugin-api } + +## State Store provider + +A `DurableStateStoreProvider` needs to be implemented to be able to create the plugin itself: + +Scala +: @@snip [MyStateStore.scala](/docs/src/main/scala/docs/persistence/state/MyStateStore.scala) { #plugin-provider } + +Java +: @@snip [MyJavaStateStoreProvider.java](/docs/src/main/java/docs/persistence/state/MyJavaStateStoreProvider.java) { #plugin-provider } + +## Configure the State Store + +A durable state store plugin can be activated with the following minimal configuration: + +Scala +: @@snip [PersistenceStatePluginDocSpec.scala](/docs/src/test/scala/docs/persistence/state/PersistenceStatePluginDocSpec.scala) { #plugin-config-scala } + +Java +: @@snip [PersistenceStatePluginDocSpec.scala](/docs/src/test/scala/docs/persistence/state/PersistenceStatePluginDocSpec.scala) { #plugin-config-java } diff --git a/docs/src/main/paradox/typed/index-persistence-durable-state.md b/docs/src/main/paradox/typed/index-persistence-durable-state.md index f96fe1eb92..aff43ca692 100644 --- a/docs/src/main/paradox/typed/index-persistence-durable-state.md +++ b/docs/src/main/paradox/typed/index-persistence-durable-state.md @@ -12,5 +12,6 @@ project.description: Durable state with Apache Pekko Persistence enables actors * [persistence-style](durable-state/persistence-style.md) * [cqrs](durable-state/cqrs.md) * [persistence-query](../durable-state/persistence-query.md) +* [state-store-plugin](../durable-state/state-store-plugin.md) @@@ diff --git a/docs/src/main/scala/docs/persistence/state/MyStateStore.scala b/docs/src/main/scala/docs/persistence/state/MyStateStore.scala new file mode 100644 index 0000000000..3d268462f8 --- /dev/null +++ b/docs/src/main/scala/docs/persistence/state/MyStateStore.scala @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * license agreements; and to You under the Apache License, version 2.0: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * This file is part of the Apache Pekko project, which was derived from Akka. + */ + +/* + * Copyright (C) 2023 Lightbend Inc. <https://www.lightbend.com> + */ + +package docs.persistence.state + +import org.apache.pekko.Done +import org.apache.pekko.actor.ExtendedActorSystem +import org.apache.pekko.persistence.state.{ DurableStateStoreProvider, DurableStateStoreRegistry } +import org.apache.pekko.persistence.state.scaladsl.{ DurableStateStore, DurableStateUpdateStore, GetObjectResult } +import com.typesafe.config.Config +import org.apache.pekko.persistence.state.javadsl.{ DurableStateStore => JDurableStateStore } +import scala.concurrent.Future + +//#plugin-provider +class MyStateStoreProvider(system: ExtendedActorSystem, config: Config, cfgPath: String) + extends DurableStateStoreProvider { + + /** + * The `DurableStateStore` implementation for the Scala API. + * This corresponds to the instance that is returned by [[DurableStateStoreRegistry#durableStateStoreFor]]. + */ + override def scaladslDurableStateStore(): DurableStateStore[Any] = new MyStateStore(system, config, cfgPath) + + /** + * The `DurableStateStore` implementation for the Java API. + * This corresponds to the instance that is returned by [[DurableStateStoreRegistry#getDurableStateStoreFor]]. + */ + override def javadslDurableStateStore(): JDurableStateStore[AnyRef] = new MyJavaStateStore(system, config, cfgPath) +} +//#plugin-provider + +//#plugin-api +class MyStateStore[A](system: ExtendedActorSystem, config: Config, cfgPath: String) extends DurableStateUpdateStore[A] { + + /** + * Will persist the latest state. If it's a new persistence id, the record will be inserted. + * + * In case of an existing persistence id, the record will be updated only if the revision + * number of the incoming record is 1 more than the already existing record. Otherwise persist will fail. + */ + override def upsertObject(persistenceId: String, revision: Long, value: A, tag: String): Future[Done] = ??? + + /** + * Deprecated. Use the deleteObject overload with revision instead. + */ + override def deleteObject(persistenceId: String): Future[Done] = deleteObject(persistenceId, 0) + + /** + * Will delete the state by setting it to the empty state and the revision number will be incremented by 1. + */ + override def deleteObject(persistenceId: String, revision: Long): Future[Done] = ??? + + /** + * Returns the current state for the given persistence id. + */ + override def getObject(persistenceId: String): Future[GetObjectResult[A]] = ??? +} +//#plugin-api diff --git a/docs/src/test/scala/docs/persistence/state/PersistenceStatePluginDocSpec.scala b/docs/src/test/scala/docs/persistence/state/PersistenceStatePluginDocSpec.scala new file mode 100644 index 0000000000..a65f50785d --- /dev/null +++ b/docs/src/test/scala/docs/persistence/state/PersistenceStatePluginDocSpec.scala @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * license agreements; and to You under the Apache License, version 2.0: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * This file is part of the Apache Pekko project, which was derived from Akka. + */ + +/* + * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com> + */ + +package docs.persistence.state + +import org.apache.pekko.Done +import org.apache.pekko.actor.{ ActorSystem, ExtendedActorSystem } + +import org.apache.pekko.persistence.state.scaladsl.DurableStateStore +import org.apache.pekko.persistence.state.{ DurableStateStoreProvider, DurableStateStoreRegistry } +import org.apache.pekko.testkit.TestKit +import com.typesafe.config._ +import docs.persistence +import org.scalatest.wordspec.AnyWordSpec + +import scala.concurrent.Future +import scala.concurrent.duration._ + +//#plugin-imports +import org.apache.pekko.persistence._ +import org.apache.pekko.persistence.state.scaladsl.DurableStateUpdateStore +import org.apache.pekko.persistence.state.scaladsl.GetObjectResult +//#plugin-imports + +class PersistenceStatePluginDocSpec extends AnyWordSpec { + + val providerConfigJava = + """ + //#plugin-config-java + # Path to the state store plugin to be used + pekko.persistence.state.plugin = "my-java-state-store" + + # My custom state store plugin + my-java-state-store { + # Class name of the plugin. + class = "docs.persistence.state.MyJavaStateStoreProvider" + } + //#plugin-config-java + """ + + val providerConfig = + """ + //#plugin-config-scala + # Path to the state store plugin to be used + pekko.persistence.state.plugin = "my-state-store" + + # My custom state store plugin + my-state-store { + # Class name of the plugin. + class = "docs.persistence.state.MyStateStoreProvider" + } + //#plugin-config-scala + """ + + "it should work for scala" in { + val system = ActorSystem("PersistenceStatePluginDocSpec", ConfigFactory.parseString(providerConfig)) + try { + Persistence(system) + DurableStateStoreRegistry(system).durableStateStoreFor[DurableStateUpdateStore[Any]]("my-state-store") + } finally { + TestKit.shutdownActorSystem(system, 10.seconds, false) + } + } + + "it should work for java" in { + val system = ActorSystem("PersistenceStatePluginDocSpec", ConfigFactory.parseString(providerConfigJava)) + try { + Persistence(system) + DurableStateStoreRegistry(system).durableStateStoreFor[DurableStateUpdateStore[Any]]("my-java-state-store") + } finally { + TestKit.shutdownActorSystem(system, 10.seconds, false) + } + } +} diff --git a/persistence/src/main/scala/org/apache/pekko/persistence/state/DurableStateStoreProvider.scala b/persistence/src/main/scala/org/apache/pekko/persistence/state/DurableStateStoreProvider.scala index b765645313..6373636aa1 100644 --- a/persistence/src/main/scala/org/apache/pekko/persistence/state/DurableStateStoreProvider.scala +++ b/persistence/src/main/scala/org/apache/pekko/persistence/state/DurableStateStoreProvider.scala @@ -24,7 +24,7 @@ package org.apache.pekko.persistence.state trait DurableStateStoreProvider { /** - * The `ReadJournal` implementation for the Scala API. + * The `DurableStateStore` implementation for the Scala API. * This corresponds to the instance that is returned by [[org.apache.pekko.persistence.state.DurableStateStoreRegistry.durableStateStoreFor DurableStateStoreRegistry#durableStateStoreFor]]. */ def scaladslDurableStateStore(): scaladsl.DurableStateStore[Any] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
