Use Netty's pooled memory allocator

Instead of the adaptative one, which is the default in 4.2.
The pooled allocator was the default in 4.1.

Author: acogoluegnes

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

@@ -256,7 +256,7 @@ public Client(ClientParameters parameters) {
       b.option(
           ChannelOption.ALLOCATOR,
           parameters.byteBufAllocator == null
-              ? ByteBufAllocator.DEFAULT
+              ? Utils.byteBufAllocator()
               : parameters.byteBufAllocator);
     }
 

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

@@ -111,6 +111,7 @@ private void maybeCompleteBatch(State<T> state, boolean increaseIfCompleted) {
         state.items = new ArrayList<>(state.batchSize);
       }
     } catch (Exception e) {
+      //      e.printStackTrace();
       LOGGER.warn("Error during dynamic batch completion: {}", e.getMessage());
     }
   }

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

@@ -422,7 +422,7 @@ static class DefaultNettyConfiguration implements NettyConfiguration {
 
     private final EnvironmentBuilder environmentBuilder;
     private EventLoopGroup eventLoopGroup;
-    private ByteBufAllocator byteBufAllocator = ByteBufAllocator.DEFAULT;
+    private ByteBufAllocator byteBufAllocator = Utils.byteBufAllocator();
     private Consumer<Channel> channelCustomizer = noOpConsumer();
     private Consumer<Bootstrap> bootstrapCustomizer = noOpConsumer();
 

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

@@ -19,6 +19,8 @@
 
 import com.rabbitmq.stream.*;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.PooledByteBufAllocator;
 import io.netty.channel.ConnectTimeoutException;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.MultiThreadIoEventLoopGroup;
@@ -415,6 +417,10 @@ static EventLoopGroup eventLoopGroup() {
     return new MultiThreadIoEventLoopGroup(NioIoHandler.newFactory());
   }
 
+  static ByteBufAllocator byteBufAllocator() {
+    return PooledByteBufAllocator.DEFAULT;
+  }
+
   /*
   class to help testing SAC on super streams
    */

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

@@ -20,7 +20,6 @@
 import com.rabbitmq.stream.impl.ServerFrameHandler.DeliverVersion1FrameHandler;
 import com.rabbitmq.stream.metrics.NoOpMetricsCollector;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Random;
@@ -51,7 +50,7 @@ public MessageBuilder messageBuilder() {
 
   ByteBuf generateFrameBuffer(
       int nbMessages, long chunkOffset, int dataSize, Iterable<byte[]> messages) {
-    ByteBuf bb = ByteBufAllocator.DEFAULT.buffer(1024);
+    ByteBuf bb = Utils.byteBufAllocator().buffer(1024);
     bb.writeShort(Utils.encodeRequestCode(Constants.COMMAND_DELIVER))
         .writeShort(Constants.VERSION_1)
         .writeByte(1) // subscription id

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

@@ -26,7 +26,6 @@
 import com.rabbitmq.stream.Message;
 import com.rabbitmq.stream.Properties;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import java.io.ByteArrayOutputStream;
@@ -149,7 +148,7 @@ public TestDesc(String description, List<Integer> sizes, int expectedCalls) {
       tests.forEach(
           test -> {
             Channel channel = Mockito.mock(Channel.class);
-            Mockito.when(channel.alloc()).thenReturn(ByteBufAllocator.DEFAULT);
+            Mockito.when(channel.alloc()).thenReturn(Utils.byteBufAllocator());
             Mockito.when(channel.writeAndFlush(Mockito.any()))
                 .thenReturn(Mockito.mock(ChannelFuture.class));
 

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

@@ -216,7 +216,7 @@ void producersAndConsumersShouldBeClosedWhenEnvironmentIsClosed(boolean lazyInit
   }
 
   @Test
-  void growShrinkResourcesWhenProducersConsumersAreOpenedAndClosed(TestInfo info) throws Exception {
+  void growShrinkResourcesWhenProducersConsumersAreOpenedAndClosed(TestInfo info) {
     int messageCount = 100;
     int streamCount = 20;
     int producersCount = ProducersCoordinator.MAX_PRODUCERS_PER_CLIENT * 3 + 10;

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

@@ -26,7 +26,6 @@
 import com.rabbitmq.stream.StreamException;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
 import com.rabbitmq.stream.impl.StreamEnvironment.LocatorNotAvailableException;
-import io.netty.buffer.ByteBufAllocator;
 import java.net.URI;
 import java.time.Duration;
 import java.util.Arrays;
@@ -94,7 +93,7 @@ Client.ClientParameters duplicate() {
             ProducersCoordinator.MAX_TRACKING_CONSUMERS_PER_CLIENT,
             ConsumersCoordinator.MAX_SUBSCRIPTIONS_PER_CLIENT,
             null,
-            ByteBufAllocator.DEFAULT,
+            Utils.byteBufAllocator(),
             false,
             type -> "locator-connection",
             cf,
@@ -163,7 +162,7 @@ void shouldTryUrisOnInitializationFailure() throws Exception {
             ProducersCoordinator.MAX_TRACKING_CONSUMERS_PER_CLIENT,
             ConsumersCoordinator.MAX_SUBSCRIPTIONS_PER_CLIENT,
             null,
-            ByteBufAllocator.DEFAULT,
+            Utils.byteBufAllocator(),
             false,
             type -> "locator-connection",
             cf,
@@ -195,7 +194,7 @@ void shouldNotOpenConnectionWhenLazyInitIsEnabled(
             ProducersCoordinator.MAX_TRACKING_CONSUMERS_PER_CLIENT,
             ConsumersCoordinator.MAX_SUBSCRIPTIONS_PER_CLIENT,
             null,
-            ByteBufAllocator.DEFAULT,
+            Utils.byteBufAllocator(),
             lazyInit,
             type -> "locator-connection",
             cf,

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

@@ -75,7 +75,7 @@ public class StreamProducerUnitTest {
   void init() {
     mocks = MockitoAnnotations.openMocks(this);
     executorService = Executors.newScheduledThreadPool(2);
-    when(channel.alloc()).thenReturn(ByteBufAllocator.DEFAULT);
+    when(channel.alloc()).thenReturn(Utils.byteBufAllocator());
     when(channel.writeAndFlush(Mockito.any())).thenReturn(channelFuture);
     when(client.allocateNoCheck(any(ByteBufAllocator.class), anyInt()))
         .thenAnswer(