Add dynamic batch option in producer builder

Author: acogoluegnes

src/docs/asciidoc/api.adoc

@@ -455,6 +455,10 @@ blocking when the limit is reached.
 |Period to send a batch of messages.
 |100 ms
 
+|`dynamicBatch`
+|Adapt batch size depending on ingress rate.
+|false
+
 |`confirmTimeout`
 |[[producer-confirm-timeout-configuration-entry]]Time before the client calls the confirm callback to signal
 outstanding unconfirmed messages timed out.

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

@@ -97,6 +97,28 @@ public interface ProducerBuilder {
    */
   ProducerBuilder batchPublishingDelay(Duration batchPublishingDelay);
 
+  /**
+   * Adapt batch size depending on ingress rate.
+   *
+   * <p>A dynamic-batch approach improves latency for low ingress rates. It can be counterproductive
+   * for sustained high ingress rates.
+   *
+   * <p>Set this flag to <code>true</code> if you want as little delay as possible before calling
+   * {@link Producer#send(Message, ConfirmationHandler)} and the message being sent to the broker.
+   *
+   * <p>Set this flag to <code>false</code> if latency is not critical for your use case and you
+   * want the highest throughput possible for both publishing and consuming.
+   *
+   * <p>Dynamic batch is not activated by default (<code>dynamicBatch = false</code>).
+   *
+   * <p>Dynamic batch is experimental.
+   *
+   * @param dynamicBatch
+   * @return this builder instance
+   * @since 0.20.0
+   */
+  ProducerBuilder dynamicBatch(boolean dynamicBatch);
+
   /**
    * The maximum number of unconfirmed outbound messages.
    *

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

@@ -102,11 +102,18 @@ public ProducerBuilder compression(Compression compression) {
     return this;
   }
 
+  @Override
   public StreamProducerBuilder batchPublishingDelay(Duration batchPublishingDelay) {
     this.batchPublishingDelay = batchPublishingDelay;
     return this;
   }
 
+  @Override
+  public ProducerBuilder dynamicBatch(boolean dynamicBatch) {
+    this.dynamicBatch = dynamicBatch;
+    return this;
+  }
+
   @Override
   public ProducerBuilder maxUnconfirmedMessages(int maxUnconfirmedMessages) {
     if (maxUnconfirmedMessages <= 0) {