Support super streams in StreamCreator

Author: acogoluegnes

src/docs/asciidoc/super-streams.adoc

@@ -58,14 +58,51 @@ When a super stream is in use, the stream Java client queries this information t
 From the application code point of view, using a super stream is mostly configuration-based.
 Some logic must also be provided to extract routing information from messages.
 
-==== Super Stream Creation
+==== Super Stream Creation and Deletion
 
-It is possible to create the topology of a super stream with any AMQP 0.9.1 library or with the https://www.rabbitmq.com/management.html[management plugin], but the `rabbitmq-streams add_super_stream` command is a handy shortcut.
-Here is how to create an invoices super stream with 3 partitions:
+It is possible to manage super streams with
+
+* the stream Java client, by using `Environment#streamCreator()` and `Environment#deleteSuperStream(String)`
+* the `add_super_stream` and `delete_super_stream` commands in `rabbitmq-streams` (CLI)
+* any AMQP 0.9.1 client library
+* the https://www.rabbitmq.com/management.html[management plugin]
+
+The stream Java client and the dedicated CLI commands are easier to use as they take care of the topology details (exchange, streams, and bindings).
+
+===== With the Client Library
+
+Here is how to create an `invoices` super stream with 5 partitions:
+
+.Creating a super stream by specifying the number of partitions
+[source,java,indent=0]
+--------
+include::{test-examples}/SuperStreamUsage.java[tag=creation-partitions]
+--------
+
+The super stream partitions will be `invoices-0`, `invoices-1`, ..., `invoices-5`.
+We use this kind of topology when routing keys of outbound messages are hashed to pick the partition to publish them to.
+This way, if the routing key is the customer ID of the invoice, all the invoices for a given customer end up in the same partition, and they can be processed in the publishing order.
+
+It is also possible to specify routing keys when creating a super stream:
+
+.Creating a super stream by specifying the routing keys
+[source,java,indent=0]
+--------
+include::{test-examples}/SuperStreamUsage.java[tag=creation-routing-keys]
+--------
+
+The super stream partitions will be `invoices-amer`, `invoices-emea` and `invoices-apac` in this case.
+
+Using one type of topology or the other depends on the use cases, especially how messages are processed.
+See the next sections on publishing and consuming to find out more.
+
+===== With the CLI
+
+Here is how to create an `invoices` super stream with 5 partitions:
 
 .Creating a super stream from the CLI
 ----
-rabbitmq-streams add_super_stream invoices --partitions 3
+rabbitmq-streams add_super_stream invoices --partitions 5
 ----
 
 Use `rabbitmq-streams add_super_stream --help` to learn more about the command.

src/main/java/com/rabbitmq/stream/Constants.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").

src/main/java/com/rabbitmq/stream/Environment.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -58,9 +58,19 @@ static EnvironmentBuilder builder() {
    * Delete a stream
    *
    * @param stream the stream to delete
+   * @since 0.15.0
    */
   void deleteStream(String stream);
 
+  /**
+   * Delete a super stream.
+   *
+   * <p>Requires RabbitMQ 3.13.0 or more.
+   *
+   * @param superStream the super stream to delete
+   */
+  void deleteSuperStream(String superStream);
+
   /**
    * Query statistics on a stream.
    *

src/main/java/com/rabbitmq/stream/StreamCreator.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -23,13 +23,24 @@ public interface StreamCreator {
   ByteCapacity MAX_SEGMENT_SIZE = ByteCapacity.from("3GB");
 
   /**
-   * The name of the stream
+   * The name of the stream.
+   *
+   * <p>Alias for {@link #name(String)}.
    *
    * @param stream
    * @return this creator instance
    */
   StreamCreator stream(String stream);
 
+  /**
+   * The name of the (super) stream.
+   *
+   * @param name
+   * @return this creator instance
+   * @since 0.15.0
+   */
+  StreamCreator name(String name);
+
   /**
    * The maximum size of the stream before it gets truncated.
    *
@@ -80,6 +91,16 @@ public interface StreamCreator {
    */
   StreamCreator filterSize(int size);
 
+  /**
+   * Configure the super stream to create.
+   *
+   * <p>Requires RabbitMQ 3.13.0 or more.
+   *
+   * @return the super stream configuration
+   * @since 0.15.0
+   */
+  SuperStreamConfiguration superStream();
+
   /**
    * Create the stream.
    *
@@ -142,4 +163,39 @@ public String value() {
       return this.value;
     }
   }
+
+  /**
+   * Super stream configuration.
+   *
+   * @since 0.15.0
+   */
+  interface SuperStreamConfiguration {
+
+    /**
+     * The number of partitions of the super stream.
+     *
+     * <p>Mutually exclusive with {@link #routingKeys(String...)}. Default is 3.
+     *
+     * @param partitions
+     * @return this super stream configuration instance
+     */
+    SuperStreamConfiguration partitions(int partitions);
+
+    /**
+     * The routing keys to use when declaring the super stream partitions.
+     *
+     * <p>Mutually exclusive with {@link #partitions(int)}. Default is null.
+     *
+     * @param routingKeys
+     * @return this super stream configuration instance
+     */
+    SuperStreamConfiguration routingKeys(String... routingKeys);
+
+    /**
+     * Go back to the creator.
+     *
+     * @return the stream creator
+     */
+    StreamCreator creator();
+  }
 }

src/main/java/com/rabbitmq/stream/impl/StreamEnvironment.java

@@ -460,6 +460,22 @@ public void deleteStream(String stream) {
     }
   }
 
+  @Override
+  public void deleteSuperStream(String superStream) {
+    checkNotClosed();
+    this.maybeInitializeLocator();
+    Client.Response response = this.locator().deleteSuperStream(superStream);
+    if (!response.isOk()) {
+      throw new StreamException(
+          "Error while deleting super stream "
+              + superStream
+              + " ("
+              + formatConstant(response.getResponseCode())
+              + ")",
+          response.getResponseCode());
+    }
+  }
+
   @Override
   public StreamStats queryStreamStats(String stream) {
     checkNotClosed();

src/main/java/com/rabbitmq/stream/impl/StreamStreamCreator.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
@@ -15,27 +15,39 @@
 
 import static com.rabbitmq.stream.impl.Utils.formatConstant;
 import static com.rabbitmq.stream.impl.Utils.namedFunction;
+import static java.util.stream.Collectors.toList;
 
 import com.rabbitmq.stream.ByteCapacity;
 import com.rabbitmq.stream.Constants;
 import com.rabbitmq.stream.StreamCreator;
 import com.rabbitmq.stream.StreamException;
 import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Function;
+import java.util.stream.IntStream;
 
 class StreamStreamCreator implements StreamCreator {
 
   private final StreamEnvironment environment;
   private final Client.StreamParametersBuilder streamParametersBuilder =
       new Client.StreamParametersBuilder().leaderLocator(LeaderLocator.LEAST_LEADERS);
-  private String stream;
+  private String name;
+  private DefaultSuperStreamConfiguration superStreamConfiguration;
 
   StreamStreamCreator(StreamEnvironment environment) {
     this.environment = environment;
   }
 
   @Override
   public StreamCreator stream(String stream) {
-    this.stream = stream;
+    this.name = stream;
+    return this;
+  }
+
+  @Override
+  public StreamCreator name(String name) {
+    this.name = name;
     return this;
   }
 
@@ -73,27 +85,104 @@ public StreamCreator filterSize(int size) {
     return this;
   }
 
+  @Override
+  public SuperStreamConfiguration superStream() {
+    if (this.superStreamConfiguration == null) {
+      this.superStreamConfiguration = new DefaultSuperStreamConfiguration(this);
+    }
+    return this.superStreamConfiguration;
+  }
+
   @Override
   public void create() {
-    if (stream == null) {
-      throw new IllegalArgumentException("Stream cannot be null");
+    if (name == null) {
+      throw new IllegalArgumentException("Stream name cannot be null");
+    }
+    Function<Client, Client.Response> function;
+    boolean superStream = this.superStreamConfiguration != null;
+    if (superStream) {
+      List<String> partitions, routingKeys;
+      if (this.superStreamConfiguration.routingKeys == null) {
+        partitions =
+            IntStream.range(0, this.superStreamConfiguration.partitions)
+                .mapToObj(i -> this.name + "-" + i)
+                .collect(toList());
+        routingKeys =
+            IntStream.range(0, this.superStreamConfiguration.partitions)
+                .mapToObj(String::valueOf)
+                .collect(toList());
+      } else {
+        partitions =
+            this.superStreamConfiguration.routingKeys.stream()
+                .map(rk -> this.name + "-" + rk)
+                .collect(toList());
+        routingKeys = this.superStreamConfiguration.routingKeys;
+      }
+      function =
+          namedFunction(
+              c ->
+                  c.createSuperStream(
+                      this.name, partitions, routingKeys, streamParametersBuilder.build()),
+              "Creation of super stream '%s'",
+              this.name);
+    } else {
+      function =
+          namedFunction(
+              c -> c.create(name, streamParametersBuilder.build()),
+              "Creation of stream '%s'",
+              this.name);
     }
     this.environment.maybeInitializeLocator();
-    Client.Response response =
-        environment.locatorOperation(
-            namedFunction(
-                c -> c.create(stream, streamParametersBuilder.build()),
-                "Creation of stream '%s'",
-                this.stream));
+    Client.Response response = environment.locatorOperation(function);
     if (!response.isOk()
         && response.getResponseCode() != Constants.RESPONSE_CODE_STREAM_ALREADY_EXISTS) {
+      String label = superStream ? "super stream" : "stream";
       throw new StreamException(
-          "Error while creating stream '"
-              + stream
+          "Error while creating "
+              + label
+              + " '"
+              + name
               + "' ("
               + formatConstant(response.getResponseCode())
               + ")",
           response.getResponseCode());
     }
   }
+
+  private static class DefaultSuperStreamConfiguration implements SuperStreamConfiguration {
+
+    private final StreamCreator creator;
+
+    private int partitions = 3;
+    private List<String> routingKeys = null;
+
+    private DefaultSuperStreamConfiguration(StreamCreator creator) {
+      this.creator = creator;
+    }
+
+    @Override
+    public SuperStreamConfiguration partitions(int partitions) {
+      if (partitions <= 0) {
+        throw new IllegalArgumentException("The number of partitions must be greater than 0");
+      }
+      this.partitions = partitions;
+      this.routingKeys = null;
+      return this;
+    }
+
+    @Override
+    public SuperStreamConfiguration routingKeys(String... routingKeys) {
+      if (routingKeys == null || routingKeys.length == 0) {
+        throw new IllegalArgumentException("There must be at least 1 routing key");
+      }
+      this.routingKeys = Arrays.asList(routingKeys);
+      this.partitions = -1;
+      return this;
+    }
+
+    @Override
+    public StreamCreator creator() {
+      return this.creator;
+    }
+  }
 }

src/test/java/com/rabbitmq/stream/Host.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2022 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2023 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the
 // Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").

src/test/java/com/rabbitmq/stream/docs/SuperStreamUsage.java

@@ -26,6 +26,22 @@
 
 public class SuperStreamUsage {
 
+  void creation() {
+    Environment environment = Environment.builder().build();
+    // tag::creation-partitions[]
+    environment.streamCreator().name("invoices")
+        .superStream()
+        .partitions(5).creator()
+        .create();
+    // end::creation-partitions[]
+    // tag::creation-routing-keys[]
+    environment.streamCreator().name("invoices")
+        .superStream()
+        .routingKeys("amer", "emea", "apac").creator()
+        .create();
+    // end::creation-routing-keys[]
+  }
+
     void producerSimple() {
         Environment environment = Environment.builder().build();
         // tag::producer-simple[]