modify test infrastructure to detect buffer leaks, and fix one leaky test

Signed-off-by: Ludovic Orban <[email protected]>

Author: lorban

src/test/java/org/eclipse/jetty/reactive/client/AbstractTest.java

@@ -16,12 +16,14 @@
 package org.eclipse.jetty.reactive.client;
 
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.client.HttpClientTransport;
 import org.eclipse.jetty.client.transport.HttpClientTransportOverHTTP;
 import org.eclipse.jetty.http2.client.HTTP2Client;
 import org.eclipse.jetty.http2.client.transport.HttpClientTransportOverHTTP2;
 import org.eclipse.jetty.http2.server.HTTP2CServerConnectionFactory;
+import org.eclipse.jetty.io.ArrayByteBufferPool;
 import org.eclipse.jetty.io.ClientConnector;
 import org.eclipse.jetty.server.ConnectionFactory;
 import org.eclipse.jetty.server.Handler;
@@ -31,10 +33,14 @@
 import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.util.component.LifeCycle;
 import org.eclipse.jetty.util.thread.QueuedThreadPool;
+import org.hamcrest.MatcherAssert;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestInfo;
 
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.is;
+
 public class AbstractTest {
     public static void printTestName(TestInfo testInfo) {
         System.err.printf("Running %s%n", testInfo.getTestMethod()
@@ -50,6 +56,8 @@ public static List<String> protocols() {
     private HttpClient httpClient;
     protected Server server;
     private ServerConnector connector;
+    private ArrayByteBufferPool.Tracking serverBufferPool;
+    private ArrayByteBufferPool.Tracking clientBufferPool;
 
     @BeforeEach
     public void before(TestInfo testInfo) {
@@ -59,19 +67,22 @@ public void before(TestInfo testInfo) {
     public void prepare(String protocol, Handler handler) throws Exception {
         QueuedThreadPool serverThreads = new QueuedThreadPool();
         serverThreads.setName("server");
-        server = new Server(serverThreads);
+        serverBufferPool = new ArrayByteBufferPool.Tracking();
+        server = new Server(serverThreads, null, serverBufferPool);
         connector = new ServerConnector(server, 1, 1, createServerConnectionFactory(protocol));
         server.addConnector(connector);
         server.setHandler(handler);
         server.start();
 
         QueuedThreadPool clientThreads = new QueuedThreadPool();
         clientThreads.setName("client");
+        clientBufferPool = new ArrayByteBufferPool.Tracking();
         ClientConnector clientConnector = new ClientConnector();
         clientConnector.setExecutor(clientThreads);
         clientConnector.setSelectors(1);
         httpClient = new HttpClient(createClientTransport(clientConnector, protocol));
         httpClient.setExecutor(clientThreads);
+        httpClient.setByteBufferPool(clientBufferPool);
         httpClient.start();
     }
 
@@ -91,8 +102,16 @@ private HttpClientTransport createClientTransport(ClientConnector clientConnecto
 
     @AfterEach
     public void dispose() {
-        LifeCycle.stop(httpClient);
-        LifeCycle.stop(server);
+        try
+        {
+            await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> MatcherAssert.assertThat("Client leaks: " + clientBufferPool.dumpLeaks(), clientBufferPool.getLeaks().size(), is(0)));
+            await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> MatcherAssert.assertThat("Server leaks: " + serverBufferPool.dumpLeaks(), serverBufferPool.getLeaks().size(), is(0)));
+        }
+        finally
+        {
+            LifeCycle.stop(httpClient);
+            LifeCycle.stop(server);
+        }
     }
 
     protected HttpClient httpClient() {

src/test/java/org/eclipse/jetty/reactive/client/HTTP2Test.java

@@ -32,17 +32,20 @@
 import org.eclipse.jetty.http2.client.transport.HttpClientTransportOverHTTP2;
 import org.eclipse.jetty.http2.frames.HeadersFrame;
 import org.eclipse.jetty.http2.server.RawHTTP2ServerConnectionFactory;
+import org.eclipse.jetty.io.ArrayByteBufferPool;
 import org.eclipse.jetty.io.ClientConnector;
 import org.eclipse.jetty.server.Server;
 import org.eclipse.jetty.server.ServerConnector;
 import org.eclipse.jetty.util.component.LifeCycle;
 import org.eclipse.jetty.util.thread.QueuedThreadPool;
+import org.hamcrest.MatcherAssert;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
+import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -51,29 +54,42 @@ public class HTTP2Test {
     private Server server;
     private ServerConnector connector;
     private HttpClient httpClient;
+    private ArrayByteBufferPool.Tracking serverBufferPool;
+    private ArrayByteBufferPool.Tracking clientBufferPool;
 
     private void start(ServerSessionListener listener) throws Exception {
         QueuedThreadPool serverThreads = new QueuedThreadPool();
         serverThreads.setName("server");
-        server = new Server(serverThreads);
+        serverBufferPool = new ArrayByteBufferPool.Tracking();
+        server = new Server(serverThreads, null, serverBufferPool);
         RawHTTP2ServerConnectionFactory h2c = new RawHTTP2ServerConnectionFactory(listener);
         connector = new ServerConnector(server, 1, 1, h2c);
         server.addConnector(connector);
         server.start();
 
         QueuedThreadPool clientThreads = new QueuedThreadPool();
         clientThreads.setName("client");
+        clientBufferPool = new ArrayByteBufferPool.Tracking();
         ClientConnector clientConnector = new ClientConnector();
         clientConnector.setExecutor(clientThreads);
         clientConnector.setSelectors(1);
         httpClient = new HttpClient(new HttpClientTransportOverHTTP2(new HTTP2Client(clientConnector)));
+        httpClient.setByteBufferPool(clientBufferPool);
         httpClient.start();
     }
 
     @AfterEach
     public void dispose() {
-        LifeCycle.stop(httpClient);
-        LifeCycle.stop(server);
+        try
+        {
+            MatcherAssert.assertThat("Client leaks: " + clientBufferPool.dumpLeaks(), clientBufferPool.getLeaks().size(), is(0));
+            MatcherAssert.assertThat("Server leaks: " + serverBufferPool.dumpLeaks(), serverBufferPool.getLeaks().size(), is(0));
+        }
+        finally
+        {
+            LifeCycle.stop(httpClient);
+            LifeCycle.stop(server);
+        }
     }
 
     @Test

src/test/java/org/eclipse/jetty/reactive/client/RxJavaTest.java

@@ -439,6 +439,7 @@ public void onComplete() {
         // There should be 1 chunk only.
         await().during(1, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS).until(chunks::get, is(1));
         Content.Chunk chunk = await().atMost(5, TimeUnit.SECONDS).until(() -> subscriber.chunk, notNullValue());
+        chunk.release();
         subscriber.chunk = null;
         assertEquals("hello", UTF_8.decode(chunk.getByteBuffer()).toString());
 
@@ -448,6 +449,7 @@ public void onComplete() {
         // Demand 1 more chunk.
         subscriber.subscription.request(1);
         chunk = await().atMost(5, TimeUnit.SECONDS).until(() -> subscriber.chunk, notNullValue());
+        chunk.release();
         subscriber.chunk = null;
         assertEquals("world", UTF_8.decode(chunk.getByteBuffer()).toString());