Add producer names option to stream perf test

If producer names are set, message de-duplication will be enabled.

The default behaviour does not change: producer names are empty and
de-duplication is therefore disabled.

Author: ansd

src/docs/asciidoc/api.adoc

@@ -346,7 +346,7 @@ The following table sums up the main settings to create a `Producer`:
 
 |`name`
 |The logical name of the producer. Specify a name to enable
-<<outbound-message-de-deduplication,message de-deduplication>>.
+<<outbound-message-de-duplication,message de-duplication>>.
 |`null` (no de-duplication)
 
 |`batchSize`
@@ -475,8 +475,8 @@ type system. It provides good interoperability, which allows streams
 to be accessed as AMQP 0-9-1 queues, without data loss.
 ====
 
-[[outbound-message-de-deduplication]]
-===== Message De-deduplication
+[[outbound-message-de-duplication]]
+===== Message De-duplication
 
 RabbitMQ Stream provides publisher confirms to avoid losing messages: once
 the broker has persisted a message it sends a confirmation for this message.

src/docs/asciidoc/overview.adoc

@@ -49,7 +49,7 @@ you need to use an AMQP 0-9-1 client library.
 RabbitMQ stream provides at-least-once guarantees thanks to the
 publisher confirm mechanism, which is supported by the stream Java client.
 
-Message <<api.adoc#outbound-message-de-deduplication,de-duplication>>
+Message <<api.adoc#outbound-message-de-duplication,de-duplication>>
 is also supported on the publisher side.
 
 [[stream-client-overview]]

src/docs/asciidoc/performance-tool.adoc

@@ -404,6 +404,7 @@ The following command shows how to store the offset every 100,000 messages:
 java -jar stream-perf-test.jar --store-every 100000
 ----
 
+[[consumer-names]]
 ===== Consumer Names
 
 When using `--store-every` (see above) for <<api.adoc#consumer-offset-tracking, offset tracking>>,
@@ -448,6 +449,13 @@ force the offset they start consuming from. With consumer names that do not chan
 tracking consumers would ignore the specified offset and would start where they left off
 (this is the purpose of offset tracking).
 
+===== Producer Names
+You can use the `--producer-names` option to set the producer names pattern and therefore
+enable <<api.adoc#outbound-message-de-duplication, message de-duplication>> (using the default
+publishing sequence starting at 0 and incremented for each message).
+The same naming options apply as above in <<api.adoc#consumer-names, consumer names>> with the only
+difference that the default pattern is empty (i.e. no de-duplication).
+
 === Building the Performance Tool
 
 To build the uber JAR:

src/main/java/com/rabbitmq/stream/perf/StreamPerfTest.java

@@ -30,6 +30,7 @@
 import com.rabbitmq.stream.EnvironmentBuilder.TlsConfiguration;
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.Producer;
+import com.rabbitmq.stream.ProducerBuilder;
 import com.rabbitmq.stream.StreamCreator;
 import com.rabbitmq.stream.StreamCreator.LeaderLocator;
 import com.rabbitmq.stream.StreamException;
@@ -258,6 +259,16 @@ public class StreamPerfTest implements Callable<Integer> {
       converter = Utils.OneTo255RangeIntegerTypeConverter.class)
   private int producersByConnection;
 
+  @CommandLine.Option(
+      names = {"--producer-names", "-pn"},
+      description =
+          "naming strategy for producer names. Valid values are 'uuid' or a pattern with "
+              + "stream name and producer index as arguments. "
+              + "If set, a publishing ID is automatically assigned to each outbound message.",
+      defaultValue = "",
+      converter = Utils.NameStrategyConverter.class)
+  private BiFunction<String, Integer, String> producerNameStrategy;
+
   @CommandLine.Option(
       names = {"--tracking-consumers-by-connection", "-ccbc"},
       description = "number of tracking consumers by connection. Value must be between 1 and 255.",
@@ -284,7 +295,7 @@ public class StreamPerfTest implements Callable<Integer> {
           "naming strategy for consumer names. Valid values are 'uuid' or a pattern with "
               + "stream name and consumer index as arguments.",
       defaultValue = "%s-%d",
-      converter = Utils.ConsumerNameStrategyConverter.class)
+      converter = Utils.NameStrategyConverter.class)
   private BiFunction<String, Integer, String> consumerNameStrategy;
 
   @CommandLine.Option(
@@ -597,10 +608,17 @@ public Integer call() throws Exception {
                   }
 
                   String stream = stream();
+                  ProducerBuilder producerBuilder = environment.producerBuilder();
+
+                  String producerName = this.producerNameStrategy.apply(stream, i + 1);
+                  if (producerName != "") {
+                      producerBuilder = producerBuilder
+                          .name(producerName)
+                          .confirmTimeout(Duration.ZERO);
+                  }
 
                   Producer producer =
-                      environment
-                          .producerBuilder()
+                      producerBuilder
                           .subEntrySize(this.subEntrySize)
                           .batchSize(this.batchSize)
                           .compression(

src/main/java/com/rabbitmq/stream/perf/Utils.java

@@ -151,15 +151,15 @@ public ByteCapacity convert(String value) {
     }
   }
 
-  static class ConsumerNameStrategyConverter
+  static class NameStrategyConverter
       implements CommandLine.ITypeConverter<BiFunction<String, Integer, String>> {
 
     @Override
     public BiFunction<String, Integer, String> convert(String input) {
       if ("uuid".equals(input)) {
         return (stream, index) -> UUID.randomUUID().toString();
       } else {
-        return new PatternConsumerNameStrategy(input);
+        return new PatternNameStrategy(input);
       }
     }
   }
@@ -429,11 +429,11 @@ public X509Certificate[] getAcceptedIssuers() {
     }
   }
 
-  static final class PatternConsumerNameStrategy implements BiFunction<String, Integer, String> {
+  static final class PatternNameStrategy implements BiFunction<String, Integer, String> {
 
     private final String pattern;
 
-    PatternConsumerNameStrategy(String pattern) {
+    PatternNameStrategy(String pattern) {
       this.pattern = pattern;
     }
 

src/test/java/com/rabbitmq/stream/perf/StreamPerfTestTest.java

@@ -195,6 +195,30 @@ void offsetShouldNotBeStoredWhenOptionIsNotEnabled() throws Exception {
     waitRunEnds();
   }
 
+  @Test
+  void publishingSequenceShouldBeStoredWhenProducerNamesAreSet() throws Exception {
+    Future<?> run = run(builder().producerNames("producer-%2$d-on-stream-%1$s"));
+    waitUntilStreamExists(s);
+    String producerName = "producer-1-on-stream-" + s;
+    long seq = client.queryPublisherSequence(producerName, s);
+    waitOneSecond();
+    waitAtMost(() -> client.queryPublisherSequence(producerName, s) > seq);
+    run.cancel(true);
+    waitRunEnds();
+  }
+
+  @Test
+  void publishingSequenceShouldNotBeStoredWhenProducerNamesAreNotSet() throws Exception {
+    Future<?> run = run(builder());
+    waitUntilStreamExists(s);
+    String producerName = s + "-0"; // convention
+    assertThat(client.queryPublisherSequence(producerName, s)).isZero();
+    waitOneSecond();
+    assertThat(client.queryPublisherSequence(producerName, s)).isZero();
+    run.cancel(true);
+    waitRunEnds();
+  }
+
   @Test
   @DisabledIfTlsNotEnabled
   void shouldConnectWithTls() throws Exception {
@@ -405,6 +429,11 @@ ArgumentsBuilder consumerNames(String pattern) {
       return this;
     }
 
+    ArgumentsBuilder producerNames(String pattern) {
+      arguments.put("producer-names", pattern);
+      return this;
+    }
+
     String build() {
       return this.arguments.entrySet().stream()
           .map(e -> "--" + e.getKey() + (e.getValue().isEmpty() ? "" : (" " + e.getValue())))

src/test/java/com/rabbitmq/stream/perf/UtilsTest.java

@@ -22,7 +22,7 @@
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.compression.Compression;
 import com.rabbitmq.stream.perf.Utils.CompressionTypeConverter;
-import com.rabbitmq.stream.perf.Utils.PatternConsumerNameStrategy;
+import com.rabbitmq.stream.perf.Utils.PatternNameStrategy;
 import com.rabbitmq.stream.perf.Utils.RangeTypeConverter;
 import com.rabbitmq.stream.perf.Utils.SniServerNamesConverter;
 import java.util.Arrays;
@@ -143,7 +143,7 @@ void compressionTypeConverterKo(String value) {
     "consumer-%2$d-on-stream-%1$s,consumer-2-on-stream-s1"
   })
   void consumerNameStrategy(String pattern, String expected) {
-    BiFunction<String, Integer, String> strategy = new PatternConsumerNameStrategy(pattern);
+    BiFunction<String, Integer, String> strategy = new PatternNameStrategy(pattern);
     assertThat(strategy.apply("s1", 2)).isEqualTo(expected);
   }