Back to Netty 4.2.0

Experiencing unexpected disconnections in some tests.

Author: acogoluegnes

pom.xml

@@ -50,7 +50,7 @@
     <spotless.check.skip>true</spotless.check.skip>
     <slf4j.version>1.7.36</slf4j.version>
     <logback.version>1.2.13</logback.version>
-    <netty.version>4.2.1.Final</netty.version>
+    <netty.version>4.2.0.Final</netty.version>
     <proton-j.version>0.34.1</proton-j.version>
     <metrics.version>4.2.30</metrics.version>
     <micrometer.version>1.14.6</micrometer.version>

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

@@ -41,6 +41,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -57,6 +58,7 @@
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
+import wiremock.org.checkerframework.checker.units.qual.A;
 
 @ExtendWith(TestUtils.StreamTestInfrastructureExtension.class)
 public class StreamProducerTest {
@@ -88,6 +90,67 @@ void tearDown() {
     environment.close();
   }
 
+  private static AtomicLong rate() {
+    AtomicLong count = new AtomicLong();
+    AtomicLong tick = new AtomicLong(System.nanoTime());
+
+    Executors.newSingleThreadScheduledExecutor()
+        .scheduleAtFixedRate(
+            () -> {
+              long now = System.nanoTime();
+              long before = tick.getAndSet(now);
+              long elapsed = now - before;
+              long sent = count.getAndSet(0);
+              System.out.println("Rate " + (sent * 1_000_000_000L / elapsed) + " msg/s");
+            },
+            1,
+            1,
+            TimeUnit.SECONDS);
+    return count;
+  }
+
+  @Test
+  void test() {
+    AtomicLong count = rate();
+    Producer producer = environment.producerBuilder().stream(stream)
+        .maxUnconfirmedMessages(10)
+        .build();
+
+    while(true) {
+      producer.send(producer.messageBuilder().build(), s -> { });
+      count.incrementAndGet();
+    }
+
+  }
+
+  @Test
+  void client() throws Exception {
+    int permits = 10;
+    Semaphore semaphore = new Semaphore(permits);
+    Client client = cf.get(new Client.ClientParameters().publishConfirmListener(new Client.PublishConfirmListener() {
+      @Override
+      public void handle(byte publisherId, long publishingId) {
+        semaphore.release();
+      }
+    }));
+
+    byte pubId = (byte) 0;
+    client.declarePublisher(pubId, null, stream);
+
+    AtomicLong count = rate();
+
+    List<Message> messages = IntStream.range(0, permits).mapToObj(ignored -> client
+        .messageBuilder()
+        .addData("hello".getBytes(StandardCharsets.UTF_8))
+        .build()).collect(Collectors.toList());
+    while (true) {
+      semaphore.acquire(permits);
+      client.publish(pubId, messages);
+      count.addAndGet(permits);
+    }
+
+  }
+
   @Test
   void send() throws Exception {
     int batchSize = 10;