Refine tests

Author: acogoluegnes

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

@@ -43,6 +43,7 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -57,7 +58,12 @@ public class DedupIdempotencyTest {
   void init() {
     EnvironmentBuilder environmentBuilder =
         Environment.builder().netty().eventLoopGroup(eventLoopGroup).environmentBuilder();
-    environment = environmentBuilder.build();
+    this.environment = environmentBuilder.build();
+  }
+
+  @AfterEach
+  void tearDown() {
+    this.environment.close();
   }
 
   @Test

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

@@ -39,11 +39,14 @@
 import com.rabbitmq.stream.impl.Client.OutboundEntityWriteCallback;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.UnpooledByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import java.time.Duration;
+import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -70,7 +73,7 @@ public class StreamProducerUnitTest {
   @Mock Channel channel;
   @Mock ChannelFuture channelFuture;
 
-  Set<ByteBuf> buffers = ConcurrentHashMap.newKeySet();
+  Queue<ByteBuf> buffers = new ConcurrentLinkedQueue<>();
 
   ScheduledExecutorService executorService;
   Clock clock = new Clock();
@@ -82,15 +85,16 @@ public class StreamProducerUnitTest {
   void init() {
     mocks = MockitoAnnotations.openMocks(this);
     executorService = Executors.newScheduledThreadPool(2);
-    when(channel.alloc()).thenReturn(Utils.byteBufAllocator());
+    ByteBufAllocator allocator = new UnpooledByteBufAllocator(false);
+    when(channel.alloc()).thenReturn(allocator);
     when(channel.writeAndFlush(Mockito.any())).thenReturn(channelFuture);
     when(client.allocateNoCheck(any(ByteBufAllocator.class), anyInt()))
         .thenAnswer(
             (Answer<ByteBuf>)
                 invocation -> {
-                  ByteBufAllocator allocator = invocation.getArgument(0);
+                  ByteBufAllocator alloc = invocation.getArgument(0);
                   int capacity = invocation.getArgument(1);
-                  ByteBuf buffer = allocator.buffer(capacity);
+                  ByteBuf buffer = alloc.buffer(capacity);
                   buffers.add(buffer);
                   return buffer;
                 });