docs and api

Author: patriknw

akka-javasdk/src/main/java/akka/javasdk/annotations/EnableReplicationFilter.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright (C) 2021-2024 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.javasdk.annotations;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface EnableReplicationFilter {
+}

akka-javasdk/src/main/java/akka/javasdk/eventsourcedentity/EventSourcedEntity.java

@@ -273,7 +273,7 @@ interface Builder<S, E> {
       /**
        * Change the replication filter without persisting any events.
        */
-      OnSuccessBuilder<S> replicationFilter(ReplicationFilter filter);
+      OnSuccessBuilder<S> updateReplicationFilter(ReplicationFilter filter);
 
       /**
        * Create a message reply.
@@ -336,7 +336,7 @@ interface OnSuccessBuilder<S> {
       /**
        * Change the replication filter combined with for example {@code persist} event.
        */
-      OnSuccessBuilder<S> replicationFilter(ReplicationFilter filter);
+      OnSuccessBuilder<S> updateReplicationFilter(ReplicationFilter filter);
 
     }
 

akka-javasdk/src/main/java/akka/javasdk/eventsourcedentity/ReplicationFilter.java

@@ -6,6 +6,8 @@
 
 import akka.javasdk.impl.effect.ReplicationFilterImpl;
 
+import java.util.Set;
+
 public interface ReplicationFilter {
   static ReplicationFilter empty() {
     return ReplicationFilterImpl.empty();
@@ -15,12 +17,24 @@ static ReplicationFilter includeRegion(String region) {
     return ReplicationFilterImpl.empty().addRegion(region);
   }
 
+  static ReplicationFilter includeRegions(Set<String> regions) {
+    return ReplicationFilterImpl.empty().addRegions(regions);
+  }
+
   static ReplicationFilter excludeRegion(String region) {
     return ReplicationFilterImpl.empty().removeRegion(region);
   }
 
+  static ReplicationFilter excludeRegions(Set<String> regions) {
+    return ReplicationFilterImpl.empty().removeRegions(regions);
+  }
+
   ReplicationFilter addRegion(String region);
 
+  ReplicationFilter addRegions(Set<String> regions);
+
   ReplicationFilter removeRegion(String region);
 
+  ReplicationFilter removeRegions(Set<String> regions);
+
 }

akka-javasdk/src/main/scala/akka/javasdk/impl/SdkRunner.scala

@@ -10,6 +10,7 @@ import java.lang.reflect.Method
 import java.util
 import java.util.Optional
 import java.util.concurrent.CompletionStage
+
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.concurrent.Promise
@@ -34,6 +35,7 @@ import akka.javasdk.Principals
 import akka.javasdk.ServiceSetup
 import akka.javasdk.Tracing
 import akka.javasdk.annotations.ComponentId
+import akka.javasdk.annotations.EnableReplicationFilter
 import akka.javasdk.annotations.GrpcEndpoint
 import akka.javasdk.annotations.Setup
 import akka.javasdk.annotations.http.HttpEndpoint
@@ -504,8 +506,7 @@ private final class Sdk(
             readOnlyCommandNames,
             instanceFactory,
             keyValue = false,
-            replicationFilterEnabled = true
-          ) // FIXME how shall we enable it? Can we solve it without enable flag?
+            replicationFilterEnabled = clz.hasAnnotation[EnableReplicationFilter])
 
       case clz if classOf[KeyValueEntity[_]].isAssignableFrom(clz) =>
         val componentId = clz.getAnnotation(classOf[ComponentId]).value
@@ -543,8 +544,7 @@ private final class Sdk(
             readOnlyCommandNames,
             instanceFactory,
             keyValue = true,
-            replicationFilterEnabled = false
-          ) // FIXME KVE replication filter
+            replicationFilterEnabled = clz.hasAnnotation[EnableReplicationFilter])
 
       case clz if Reflect.isWorkflow(clz) =>
         val componentId = clz.getAnnotation(classOf[ComponentId]).value

akka-javasdk/src/main/scala/akka/javasdk/impl/eventsourcedentity/EventSourcedEntityEffectImpl.scala

@@ -110,7 +110,7 @@ private[javasdk] class EventSourcedEntityEffectImpl[S, E]
     this.asInstanceOf[EventSourcedEntityEffectImpl[T, E]]
   }
 
-  override def replicationFilter(filter: ReplicationFilter): OnSuccessBuilder[S] = {
+  override def updateReplicationFilter(filter: ReplicationFilter): OnSuccessBuilder[S] = {
     _replicationFilter = filter.asInstanceOf[ReplicationFilterImpl]
     this
   }

akka-javasdk/src/main/scala/akka/javasdk/impl/eventsourcedentity/EventSourcedEntityImpl.scala

@@ -12,6 +12,7 @@ import scala.util.control.NonFatal
 import akka.annotation.InternalApi
 import akka.javasdk.Metadata
 import akka.javasdk.Tracing
+import akka.javasdk.annotations.EnableReplicationFilter
 import akka.javasdk.eventsourcedentity.CommandContext
 import akka.javasdk.eventsourcedentity.EventContext
 import akka.javasdk.eventsourcedentity.EventSourcedEntity
@@ -26,6 +27,7 @@ import akka.javasdk.impl.MetadataImpl
 import akka.javasdk.impl.effect.ErrorReplyImpl
 import akka.javasdk.impl.effect.MessageReplyImpl
 import akka.javasdk.impl.effect.NoSecondaryEffectImpl
+import akka.javasdk.impl.effect.ReplicationFilterImpl
 import akka.javasdk.impl.eventsourcedentity.EventSourcedEntityEffectImpl.EmitEvents
 import akka.javasdk.impl.eventsourcedentity.EventSourcedEntityEffectImpl.NoPrimaryEffect
 import akka.javasdk.impl.serialization.JsonSerializer
@@ -150,6 +152,11 @@ private[impl] final class EventSourcedEntityImpl[S, E, ES <: EventSourcedEntity[
         }
       }
 
+      if ((commandEffect.replFilter ne ReplicationFilterImpl.empty) && !isReplicationFilterEnabled) {
+        throw new IllegalStateException(
+          "To use replication filters the EventSourcedEntity class must be annotated with @EnableReplicationFilter.")
+      }
+
       var currentSequence = command.sequenceNumber
       commandEffect.primaryEffect match {
         case EmitEvents(events, deleteEntity) =>
@@ -240,4 +247,9 @@ private[impl] final class EventSourcedEntityImpl[S, E, ES <: EventSourcedEntity[
 
   override def stateFromBytes(pb: BytesPayload): SpiEventSourcedEntity.State =
     serializer.fromBytes(entityStateType, pb).asInstanceOf[SpiEventSourcedEntity.State]
+
+  private def isReplicationFilterEnabled: Boolean = {
+    import akka.javasdk.impl.reflection.Reflect.Syntax.AnnotatedElementOps
+    entity.getClass.hasAnnotation[EnableReplicationFilter]
+  }
 }

akka-javasdk/src/main/scala/akka/javasdk/impl/eventsourcedentity/ReplicationFilterImpl.scala

@@ -2,7 +2,11 @@
  * Copyright (C) 2021-2024 Lightbend Inc. <https://www.lightbend.com>
  */
 
-package akka.javasdk.impl.effect
+package akka.javasdk.impl.eventsourcedentity
+
+import java.util
+
+import scala.jdk.CollectionConverters._
 
 import akka.annotation.InternalApi
 import akka.javasdk.eventsourcedentity.ReplicationFilter
@@ -25,9 +29,15 @@ private[javasdk] final case class ReplicationFilterImpl(addRegions: Set[String],
   override def addRegion(region: String): ReplicationFilterImpl =
     copy(addRegions = addRegions + region)
 
+  override def addRegions(regions: util.Set[String]): ReplicationFilter =
+    copy(addRegions = addRegions.union(regions.asScala))
+
   override def removeRegion(region: String): ReplicationFilterImpl =
     copy(removeRegions = removeRegions + region)
 
+  override def removeRegions(regions: util.Set[String]): ReplicationFilter =
+    copy(removeRegions = removeRegions.union(regions.asScala))
+
   def toSpi: SpiEventSourcedEntity.ChangeReplicationFilter =
     new SpiEventSourcedEntity.ChangeReplicationFilter(addRegions, removeRegions)
 

docs/src/modules/java/pages/event-sourced-entities.adoc

@@ -179,6 +179,20 @@ When the Event Sourced Entity is loaded again, the snapshot will be loaded befor
 [#_replication]
 include::partial$mutli-region-replication.adoc[]
 
+=== Replication filters
+
+Events are by default replicated to all regions that have been enabled for the service. For regulatory reasons or as cost optimization it is possible to filter which regions that participate in the replication for a specific entity. This can be changed at runtime by the entity itself.
+
+[source,java]
+.{sample-base-url}/doc-snippets/src/main/java/com/example/replicationfilter/ShoppingCartEntity.java[ShoppingCartEntity.java]
+----
+include::example$doc-snippets/src/main/java/com/example/replicationfilter/ShoppingCartEntity.java[tag=replication-filter]
+----
+<1> Enable the replication filter feature by adding the `@EnableReplicationFilter annotation.
+<2> Define the replication filter with the `updateReplicationFilter` effect.
+
+After enabling replication filter the entity is still replicated to all regions until the regions are defined with the `updateReplicationFilter` effect. This effect can be combined with persisting events or used without additional events. The filter can only be updated from the primary region of the entity, or it will become the primary if using `request-region` primary selection strategy. The filter is durable for the specific entity instance and can be changed without deploying a new version.
+
 == Side Effects
 
 An entity doesn't perform any external side effects aside from persisting events and replying to the request. Side effects can be handled from the Workflow, Consumer, or Endpoint components that are calling the entity.

samples/doc-snippets/src/main/java/com/example/replicationfilter/ShoppingCartEntity.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) 2024 Lightbend Inc. <https://www.lightbend.com>
+ */
+package com.example.replicationfilter;
+
+// tag::replication-filter[]
+import akka.Done;
+import akka.javasdk.annotations.ComponentId;
+import akka.javasdk.annotations.EnableReplicationFilter;
+import akka.javasdk.eventsourcedentity.EventSourcedEntity;
+import akka.javasdk.eventsourcedentity.ReplicationFilter;
+
+@ComponentId("shopping-cart")
+@EnableReplicationFilter // <1>
+public class ShoppingCartEntity extends EventSourcedEntity<ShoppingCart, ShoppingCartEvent> {
+
+  public Effect<Done> replicateTo(String region) {
+    return effects()
+        .updateReplicationFilter(ReplicationFilter.includeRegion(region)) // <2>
+        .thenReply(__ -> Done.getInstance());
+  }
+
+  // end::replication-filter[]
+  @Override
+  public ShoppingCart applyEvent(ShoppingCartEvent event) {
+    return currentState();
+  }
+  // tag::replication-filter[]
+}
+// end::replication-filter[]
+
+record ShoppingCart(String id) {}
+
+interface ShoppingCartEvent {}

samples/shopping-cart-quickstart/src/main/java/shoppingcart/application/ShoppingCartEntity.java

@@ -3,6 +3,7 @@
 
 import akka.Done;
 import akka.javasdk.annotations.ComponentId;
+import akka.javasdk.annotations.EnableReplicationFilter;
 import akka.javasdk.eventsourcedentity.EventSourcedEntity;
 import akka.javasdk.eventsourcedentity.EventSourcedEntityContext;
 import akka.javasdk.eventsourcedentity.ReplicationFilter;
@@ -19,6 +20,7 @@
 // tag::all[]
 // tag::class[]
 @ComponentId("shopping-cart") // <2>
+@EnableReplicationFilter
 public class ShoppingCartEntity extends EventSourcedEntity<ShoppingCart, ShoppingCartEvent> { // <1>
   // end::class[]
 
@@ -58,22 +60,18 @@ public Effect<Done> addItem(LineItem item) {
           .thenReply(newState -> Done.getInstance()); // <4>
     } else {
       logger.info("Update replication filter {}", item);
-      var filter = ReplicationFilter.empty();
-      for (String region : item.addRegions()) {
-        filter = filter.addRegion(region);
-      }
-      for (String region : item.removeRegions()) {
-        filter = filter.removeRegion(region);
-      }
+      var filter = ReplicationFilter
+          .includeRegions(item.addRegions())
+          .removeRegions(item.removeRegions());
 
       if (item.productId().isEmpty()) {
         return effects()
-            .replicationFilter(filter)
+            .updateReplicationFilter(filter)
             .thenReply(newState -> Done.getInstance());
       } else {
         return effects()
             .persist(event)
-            .replicationFilter(filter)
+            .updateReplicationFilter(filter)
             .thenReply(newState -> Done.getInstance());
 
       }