Release buffer from outstanding connection tasks

A connection can accumulate delivery frames in its dedicated executor
service. The executor service is shut down when the connection closes
and it can contain tasks that have not commenced execution. These tasks
will never get executed, so their respective buffer must be released to
avoid a memory leak.

The leak can occur with consumer churn combined witht one-consumer
connections.

Fixes #585

Author: acogoluegnes

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

@@ -352,9 +352,17 @@ public void initChannel(SocketChannel ch) {
     this.executorServiceClosing =
         Utils.makeIdempotent(
             () -> {
-              this.dispatchingExecutorService.shutdownNow();
               if (dispatchingExecutorServiceFactory == null) {
-                this.dispatchingExecutorService.shutdownNow();
+                List<Runnable> outstandingTasks = this.dispatchingExecutorService.shutdownNow();
+                for (Runnable outstandingTask : outstandingTasks) {
+                  try {
+                    outstandingTask.run();
+                  } catch (Exception e) {
+                    LOGGER.info(
+                        "Error while releasing buffer in outstanding connection tasks: {}",
+                        e.getMessage());
+                  }
+                }
               } else {
                 dispatchingExecutorServiceFactory.clientClosed(this.dispatchingExecutorService);
               }
@@ -2770,7 +2778,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
         }
       } else {
         FrameHandler frameHandler = ServerFrameHandler.lookup(commandId, version, m);
-        task = () -> frameHandler.handle(Client.this, frameSize, ctx, m);
+        //        task = () -> frameHandler.handle(Client.this, frameSize, ctx, m);
+        task = new FrameHandlerTask(frameHandler, Client.this, frameSize, ctx, m, closing);
       }
 
       if (task != null) {
@@ -2839,6 +2848,44 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
     }
   }
 
+  private static class FrameHandlerTask implements Runnable {
+
+    private final FrameHandler frameHandler;
+    private final Client client;
+    private final int frameSize;
+    private final ChannelHandlerContext ctx;
+    private final ByteBuf message;
+    private final AtomicBoolean closing;
+
+    private FrameHandlerTask(
+        FrameHandler frameHandler,
+        Client client,
+        int frameSize,
+        ChannelHandlerContext ctx,
+        ByteBuf message,
+        AtomicBoolean closing) {
+      this.frameHandler = frameHandler;
+      this.client = client;
+      this.frameSize = frameSize;
+      this.ctx = ctx;
+      this.message = message;
+      this.closing = closing;
+    }
+
+    @Override
+    public void run() {
+      if (this.closing.get()) {
+        try {
+          this.message.release();
+        } catch (Exception e) {
+          e.printStackTrace();
+        }
+      } else {
+        this.frameHandler.handle(this.client, this.frameSize, this.ctx, this.message);
+      }
+    }
+  }
+
   private <T> OutstandingRequest<T> outstandingRequest() {
     return new OutstandingRequest<>(this.rpcTimeout, this.host + ":" + this.port);
   }