Merge pull request #905 from rabbitmq/deprecate-fixed-size-batching

Deprecated producer fixed-size batching

Author: acogoluegnes

.github/workflows/test-pr.yml

@@ -30,17 +30,9 @@ jobs:
           cache: 'maven'
       - name: Start broker
         run: ci/start-broker.sh
-      - name: Test (no dynamic-batch publishing)
-        run: |
-          ./mvnw verify -Drabbitmqctl.bin=DOCKER:rabbitmq \
-            -Drabbitmq.stream.producer.dynamic.batch=false \
-            -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
-            -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
-            -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem
-      - name: Test (dynamic-batch publishing)
+      - name: Test
         run: |
           ./mvnw test -Drabbitmqctl.bin=DOCKER:rabbitmq \
-            -Drabbitmq.stream.producer.dynamic.batch=true \
             -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
             -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
             -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem

.github/workflows/test-rabbitmq-alphas.yml

@@ -40,17 +40,9 @@ jobs:
         run: ci/start-broker.sh
         env:
           RABBITMQ_IMAGE: ${{ matrix.rabbitmq-image }}
-      - name: Test (no dynamic-batch publishing)
-        run: |
-          ./mvnw verify -Drabbitmqctl.bin=DOCKER:rabbitmq \
-            -Drabbitmq.stream.producer.dynamic.batch=false \
-            -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
-            -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
-            -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem
-      - name: Test (dynamic-batch publishing)
+      - name: Test
         run: |
           ./mvnw test -Drabbitmqctl.bin=DOCKER:rabbitmq \
-          -Drabbitmq.stream.producer.dynamic.batch=true \
           -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
           -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
           -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem

.github/workflows/test-supported-java-versions.yml

@@ -30,18 +30,9 @@ jobs:
         run: ci/start-broker.sh
       - name: Display Java version
         run: ./mvnw --version
-      - name: Test (no dynamic-batch publishing)
-        run: |
-          ./mvnw verify -Drabbitmqctl.bin=DOCKER:rabbitmq \
-            -Drabbitmq.stream.producer.dynamic.batch=false \
-            -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
-            -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
-            -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem \
-            -Dnet.bytebuddy.experimental=true -Djacoco.skip=true -Dspotbugs.skip=true
-      - name: Test (dynamic-batch publishing)
+      - name: Test
         run: |
           ./mvnw test -Drabbitmqctl.bin=DOCKER:rabbitmq \
-          -Drabbitmq.stream.producer.dynamic.batch=true \
           -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
           -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
           -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem \

.github/workflows/test.yml

@@ -36,17 +36,9 @@ jobs:
           gpg-passphrase: MAVEN_GPG_PASSPHRASE
       - name: Start broker
         run: ci/start-broker.sh
-      - name: Test (no dynamic-batch publishing)
-        run: |
-          ./mvnw verify -Drabbitmqctl.bin=DOCKER:rabbitmq \
-            -Drabbitmq.stream.producer.dynamic.batch=false \
-            -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
-            -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
-            -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem
-      - name: Test (dynamic-batch publishing)
+      - name: Test
         run: |
           ./mvnw test -Drabbitmqctl.bin=DOCKER:rabbitmq \
-          -Drabbitmq.stream.producer.dynamic.batch=true \
           -Dca.certificate=./tls-gen/basic/result/ca_certificate.pem \
           -Dclient.certificate=./tls-gen/basic/result/client_$(hostname)_certificate.pem \
           -Dclient.key=./tls-gen/basic/result/client_$(hostname)_key.pem

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

@@ -110,23 +110,18 @@ public interface ProducerBuilder {
    *
    * <p>Set this flag to <code>true</code> if you want as little delay as possible between calling
    * {@link Producer#send(Message, ConfirmationHandler)} and the message being sent to the broker.
-   * Consumers should provide enough initial credits (between 5 and 10, depending on the workload),
-   * see {@link ConsumerBuilder#flow()} and {@link
-   * ConsumerBuilder.FlowConfiguration#initialCredits(int)}.
    *
    * <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. Consumers can provide 1
-   * initial credit (depending on the workload), see {@link ConsumerBuilder#flow()} and {@link
-   * ConsumerBuilder.FlowConfiguration#initialCredits(int)}.
+   * want the highest throughput possible for both publishing and consuming.
    *
    * <p>Dynamic batch is activated by default (<code>dynamicBatch = true</code>).
    *
    * @param dynamicBatch
    * @return this builder instance
    * @since 0.20.0
-   * @see ConsumerBuilder#flow()
-   * @see com.rabbitmq.stream.ConsumerBuilder.FlowConfiguration#initialCredits(int)
+   * @deprecated dynamic/adaptive will be the default and only option in future releases
    */
+  @Deprecated
   ProducerBuilder dynamicBatch(boolean dynamicBatch);
 
   /**

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

@@ -31,9 +31,6 @@
 
 class StreamProducerBuilder implements ProducerBuilder {
 
-  static final boolean DEFAULT_DYNAMIC_BATCH =
-      Boolean.parseBoolean(System.getProperty("rabbitmq.stream.producer.dynamic.batch", "true"));
-
   private final StreamEnvironment environment;
 
   private String name;
@@ -60,7 +57,7 @@ class StreamProducerBuilder implements ProducerBuilder {
 
   private Function<Message, String> filterValueExtractor;
 
-  private boolean dynamicBatch = DEFAULT_DYNAMIC_BATCH;
+  private boolean dynamicBatch = true;
 
   private final List<Resource.StateListener> listeners = new ArrayList<>();
 

src/test/java/com/rabbitmq/stream/impl/RecoveryClusterTest.java

@@ -231,8 +231,7 @@ void clusterRestart(boolean useLoadBalancer, boolean forceLeader) throws Interru
               .mapToObj(
                   i -> {
                     String s = streams.get(i % streams.size());
-                    boolean dynamicBatch = i % 2 == 0;
-                    return new ProducerState(s, dynamicBatch, environment);
+                    return new ProducerState(s, environment);
                   })
               .collect(toList());
       consumers =
@@ -369,7 +368,7 @@ void sacWithClusterRestart(boolean superStream) throws Exception {
       }
       sCreator.create();
 
-      pState = new ProducerState(s, true, superStream, environment);
+      pState = new ProducerState(s, superStream, environment);
       pState.start();
 
       Map<Integer, Boolean> consumerStatus = new ConcurrentHashMap<>();
@@ -476,14 +475,13 @@ private static class ProducerState implements AutoCloseable {
     final AtomicReference<Throwable> lastException = new AtomicReference<>();
     final AtomicReference<Instant> lastExceptionInstant = new AtomicReference<>();
 
-    private ProducerState(String stream, boolean dynamicBatch, Environment environment) {
-      this(stream, dynamicBatch, false, environment);
+    private ProducerState(String stream, Environment environment) {
+      this(stream, false, environment);
     }
 
-    private ProducerState(
-        String stream, boolean dynamicBatch, boolean superStream, Environment environment) {
+    private ProducerState(String stream, boolean superStream, Environment environment) {
       this.stream = stream;
-      ProducerBuilder builder = environment.producerBuilder().dynamicBatch(dynamicBatch);
+      ProducerBuilder builder = environment.producerBuilder();
       if (superStream) {
         builder.superStream(stream).routing(m -> m.getProperties().getMessageIdAsString());
       } else {

src/test/java/com/rabbitmq/stream/impl/StreamProducerUnitTest.java

@@ -187,7 +187,7 @@ void confirmTimeoutTaskShouldFailMessagesAfterTimeout(
             "stream",
             subEntrySize,
             10,
-            StreamProducerBuilder.DEFAULT_DYNAMIC_BATCH,
+            true,
             Compression.NONE,
             Duration.ofMillis(100),
             messageCount * 10,
@@ -239,7 +239,7 @@ void enqueueTimeoutMessageShouldBeFailedWhenEnqueueTimeoutIsReached(int subEntry
             "stream",
             subEntrySize,
             10,
-            StreamProducerBuilder.DEFAULT_DYNAMIC_BATCH,
+            true,
             Compression.NONE,
             Duration.ZERO,
             2,
@@ -281,7 +281,7 @@ void enqueueTimeoutSendingShouldBlockWhenEnqueueTimeoutIsZero(int subEntrySize)
             "stream",
             subEntrySize,
             10,
-            StreamProducerBuilder.DEFAULT_DYNAMIC_BATCH,
+            true,
             Compression.NONE,
             Duration.ZERO,
             2,