Run test suite against both versions of message accumulator

Regular and dynamic batch.

Author: acogoluegnes

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

@@ -32,11 +32,19 @@ jobs:
         run: ci/start-broker.sh
         env:
           RABBITMQ_IMAGE: ${{ matrix.rabbitmq-image }}
-      - name: Test
+      - 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)
+        run: |
+          ./mvnw verify -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
       - name: Stop broker
         run: docker stop rabbitmq && docker rm rabbitmq

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

@@ -33,12 +33,21 @@ jobs:
         run: ci/start-broker.sh
       - name: Display Java version
         run: ./mvnw --version
-      - name: Test
+      - 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)
+        run: |
+          ./mvnw verify -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 \
+            -Dnet.bytebuddy.experimental=true -Djacoco.skip=true -Dspotbugs.skip=true
       - name: Stop broker
         run: docker stop rabbitmq && docker rm rabbitmq

.github/workflows/test.yml

@@ -33,12 +33,20 @@ jobs:
           gpg-passphrase: MAVEN_GPG_PASSPHRASE
       - name: Start broker
         run: ci/start-broker.sh
-      - name: Test
+      - 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)
+        run: |
+          ./mvnw verify -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
       - name: Stop broker
         run: docker stop rabbitmq && docker rm rabbitmq
       - name: Upload Codecov report

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

@@ -86,6 +86,7 @@ class StreamProducer implements Producer {
       String stream,
       int subEntrySize,
       int batchSize,
+      boolean dynamicBatch,
       Compression compression,
       Duration batchPublishingDelay,
       int maxUnconfirmedMessages,
@@ -172,7 +173,6 @@ public int fragmentLength(Object entity) {
     if (compression != null) {
       compressionCodec = environment.compressionCodecFactory().get(compression);
     }
-    boolean dynamicBatch = true;
     this.accumulator =
         ProducerUtils.createMessageAccumulator(
             dynamicBatch,

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

@@ -27,6 +27,9 @@
 
 class StreamProducerBuilder implements ProducerBuilder {
 
+  static final boolean DEFAULT_DYNAMIC_BATCH =
+      Boolean.parseBoolean(System.getProperty("rabbitmq.stream.producer.dynamic.batch", "false"));
+
   private final StreamEnvironment environment;
 
   private String name;
@@ -53,6 +56,8 @@ class StreamProducerBuilder implements ProducerBuilder {
 
   private Function<Message, String> filterValueExtractor;
 
+  private boolean dynamicBatch = DEFAULT_DYNAMIC_BATCH;
+
   StreamProducerBuilder(StreamEnvironment environment) {
     this.environment = environment;
   }
@@ -198,6 +203,7 @@ public Producer build() {
               stream,
               subEntrySize,
               batchSize,
+              dynamicBatch,
               compression,
               batchPublishingDelay,
               maxUnconfirmedMessages,

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

@@ -176,6 +176,7 @@ void confirmTimeoutTaskShouldFailMessagesAfterTimeout(
             "stream",
             subEntrySize,
             10,
+            StreamProducerBuilder.DEFAULT_DYNAMIC_BATCH,
             Compression.NONE,
             Duration.ofMillis(100),
             messageCount * 10,
@@ -226,6 +227,7 @@ void enqueueTimeoutMessageShouldBeFailedWhenEnqueueTimeoutIsReached(int subEntry
             "stream",
             subEntrySize,
             10,
+            StreamProducerBuilder.DEFAULT_DYNAMIC_BATCH,
             Compression.NONE,
             Duration.ZERO,
             2,
@@ -266,6 +268,7 @@ void enqueueTimeoutSendingShouldBlockWhenEnqueueTimeoutIsZero(int subEntrySize)
             "stream",
             subEntrySize,
             10,
+            StreamProducerBuilder.DEFAULT_DYNAMIC_BATCH,
             Compression.NONE,
             Duration.ZERO,
             2,