Revert "Back to Netty 4.2.0"

This reverts commit 2a02f06.

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.0.Final</netty.version>
+    <netty.version>4.2.1.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,7 +41,6 @@
 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;
@@ -58,7 +57,6 @@
 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 {
@@ -90,67 +88,6 @@ 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;