feat: (java-spring) support multiple subscriptions of the same stream (#2032)

Author: aludwiko

samples/java-spring-eventsourced-customer-registry-subscriber/src/it/java/customer/api/CustomersByNameViewIntegrationTest.java

@@ -27,7 +27,7 @@ public class CustomersByNameViewIntegrationTest extends KalixIntegrationTestKitS
   private WebClient webClient;
 
   @Test
-  public void shouldReturnCustomerByName() {
+  public void shouldReturnCustomersFromViews() {
     IncomingMessages customerEvents = kalixTestKit.getStreamIncomingMessages("customer-registry", "customer_events");
 
     String bob = "bob";
@@ -36,20 +36,28 @@ public void shouldReturnCustomerByName() {
 
     customerEvents.publish(created1, "b");
     customerEvents.publish(created2, "a");
-    
+
     await()
-        .ignoreExceptions()
-        .atMost(20, TimeUnit.SECONDS)
-        .pollInterval(1, TimeUnit.SECONDS)
-        .untilAsserted(() -> {
-              Customer customer = webClient.get()
-                  .uri("/customers/by_name/" + bob)
-                  .retrieve()
-                  .bodyToFlux(Customer.class)
-                  .blockFirst(timeout);
-
-              assertThat(customer).isEqualTo(new Customer("b", created1.email(), created1.name()));
-            }
-        );
+      .ignoreExceptions()
+      .atMost(20, TimeUnit.SECONDS)
+      .pollInterval(1, TimeUnit.SECONDS)
+      .untilAsserted(() -> {
+          Customer customer = webClient.get()
+            .uri("/customers/by_name/" + bob)
+            .retrieve()
+            .bodyToFlux(Customer.class)
+            .blockFirst(timeout);
+
+          assertThat(customer).isEqualTo(new Customer("b", created1.email(), created1.name()));
+
+          Customer customer2 = webClient.get()
+            .uri("/customers/by_email/" + created2.email())
+            .retrieve()
+            .bodyToFlux(Customer.class)
+            .blockFirst(timeout);
+
+          assertThat(customer2).isEqualTo(new Customer("a", created2.email(), created2.name()));
+        }
+      );
   }
 }

samples/java-spring-eventsourced-customer-registry-subscriber/src/main/java/customer/views/CustomersByEmailView.java

@@ -0,0 +1,53 @@
+package customer.views;
+
+import kalix.javasdk.annotations.Acl;
+import kalix.javasdk.annotations.Query;
+import kalix.javasdk.annotations.Subscribe;
+import kalix.javasdk.annotations.Table;
+import kalix.javasdk.view.View;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import reactor.core.publisher.Flux;
+
+// tag::view[]
+@Table("customers_by_email")
+@Subscribe.Stream( // <1>
+  service = "customer-registry", // <2>
+  id = "customer_events", // <3>
+  consumerGroup = "customer-by-email-view" // <4>
+)
+public class CustomersByEmailView extends View<Customer> {
+  // end::view[]
+  private static final Logger logger = LoggerFactory.getLogger(CustomersByEmailView.class);
+  // tag::view[]
+
+  public UpdateEffect<Customer> onEvent( // <4>
+                                         CustomerPublicEvent.Created created) {
+    // end::view[]
+    logger.info("Received: {}", created);
+    // tag::view[]
+    var id = updateContext().eventSubject().get();
+    return effects().updateState(
+      new Customer(id, created.email(), created.name()));
+  }
+
+  public UpdateEffect<Customer> onEvent(
+    CustomerPublicEvent.NameChanged nameChanged) {
+    // end::view[]
+    logger.info("Received: {}", nameChanged);
+    // tag::view[]
+    var updated = viewState().withName(nameChanged.newName());
+    return effects().updateState(updated);
+  }
+
+  @GetMapping("/customers/by_email/{email}")
+  @Query("SELECT * FROM customers_by_email WHERE email = :email")
+  @Acl(allow = @Acl.Matcher(principal = Acl.Principal.INTERNET))
+  public Flux<Customer> findByName(@PathVariable String email) {
+    return null;
+  }
+
+}
+// end::view[]

samples/java-spring-eventsourced-customer-registry-subscriber/src/main/java/customer/views/CustomersByNameView.java

@@ -15,7 +15,8 @@
 @Table("customers_by_name")
 @Subscribe.Stream( // <1>
     service = "customer-registry", // <2>
-    id = "customer_events" // <3>
+  id = "customer_events", // <3>
+  consumerGroup = "customer-by-name-view"
 )
 public class CustomersByNameView extends View<Customer> {
   // end::view[]

sdk/java-sdk-spring/src/main/java/kalix/javasdk/annotations/Subscribe.java

@@ -152,6 +152,13 @@
      */
     String service();
 
+    /**
+     * In case you need to consume the same stream multiple times, each subscription should have a unique consumer group.
+     * <p>
+     * Changing the consumer group will lead to starting over from the beginning of the stream.
+     */
+    String consumerGroup() default "";
+
     /**
      * When there is no method in the class whose input type matches the event type:
      * <ul>

sdk/java-sdk-spring/src/main/scala/kalix/javasdk/impl/ComponentDescriptorFactory.scala

@@ -342,6 +342,7 @@ private[impl] object ComponentDescriptorFactory {
       val in = EventSource
         .newBuilder()
         .setDirect(direct)
+        .setConsumerGroup(streamAnn.consumerGroup())
 
       val eventing =
         ServiceEventing