diff --git a/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/JerseyClientHandler.java b/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/JerseyClientHandler.java
index 5531498..9ab9745 100644
--- a/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/JerseyClientHandler.java
+++ b/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/JerseyClientHandler.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -20,17 +20,14 @@
 import java.io.InputStream;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.LinkedBlockingDeque;
 
 import javax.ws.rs.core.Response;
 
 import org.glassfish.jersey.client.ClientRequest;
 import org.glassfish.jersey.client.ClientResponse;
-import org.glassfish.jersey.client.spi.AsyncConnectorCallback;
 import org.glassfish.jersey.netty.connector.internal.NettyInputStream;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.codec.http.HttpContent;
@@ -39,8 +36,6 @@
 import io.netty.handler.codec.http.HttpResponse;
 import io.netty.handler.codec.http.HttpUtil;
 import io.netty.handler.codec.http.LastHttpContent;
-import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.GenericFutureListener;
 
 /**
  * Jersey implementation of Netty channel handler.
@@ -49,19 +44,38 @@
  */
 class JerseyClientHandler extends SimpleChannelInboundHandler<HttpObject> {
 
-    private final NettyConnector connector;
-    private final LinkedBlockingDeque<ByteBuf> isList = new LinkedBlockingDeque<>();
-
-    private final AsyncConnectorCallback asyncConnectorCallback;
     private final ClientRequest jerseyRequest;
-    private final CompletableFuture future;
+    private final CompletableFuture<ClientResponse> responseAvailable;
+    private final CompletableFuture<?> responseDone;
 
-    JerseyClientHandler(NettyConnector nettyConnector, ClientRequest request,
-                        AsyncConnectorCallback callback, CompletableFuture future) {
-        this.connector = nettyConnector;
-        this.asyncConnectorCallback = callback;
+    private NettyInputStream nis;
+    private ClientResponse jerseyResponse;
+
+    JerseyClientHandler(ClientRequest request,
+                        CompletableFuture<ClientResponse> responseAvailable,
+                        CompletableFuture<?> responseDone) {
         this.jerseyRequest = request;
-        this.future = future;
+        this.responseAvailable = responseAvailable;
+        this.responseDone = responseDone;
+    }
+
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) {
+       notifyResponse();
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) {
+       // assert: no-op, if channel is closed after LastHttpContent has been consumed
+       responseDone.completeExceptionally(new IOException("Stream closed"));
+    }
+
+    protected void notifyResponse() {
+       if (jerseyResponse != null) {
+          ClientResponse cr = jerseyResponse;
+          jerseyResponse = null;
+          responseAvailable.complete(cr);
+       }
     }
 
     @Override
@@ -69,7 +83,7 @@
         if (msg instanceof HttpResponse) {
             final HttpResponse response = (HttpResponse) msg;
 
-            final ClientResponse jerseyResponse = new ClientResponse(new Response.StatusType() {
+            jerseyResponse = new ClientResponse(new Response.StatusType() {
                 @Override
                 public int getStatusCode() {
                     return response.status().code();
@@ -89,19 +103,15 @@
             for (Map.Entry<String, String> entry : response.headers().entries()) {
                 jerseyResponse.getHeaders().add(entry.getKey(), entry.getValue());
             }
-            isList.clear(); // clearing the content - possible leftover from previous request processing.
+
             // request entity handling.
             if ((response.headers().contains(HttpHeaderNames.CONTENT_LENGTH) && HttpUtil.getContentLength(response) > 0)
                     || HttpUtil.isTransferEncodingChunked(response)) {
 
-                ctx.channel().closeFuture().addListener(new GenericFutureListener<Future<? super Void>>() {
-                    @Override
-                    public void operationComplete(Future<? super Void> future) throws Exception {
-                        isList.add(Unpooled.EMPTY_BUFFER);
-                    }
-                });
+                nis = new NettyInputStream();
+                responseDone.whenComplete((_r, th) -> nis.complete(th));
 
-                jerseyResponse.setEntityStream(new NettyInputStream(isList));
+                jerseyResponse.setEntityStream(nis);
             } else {
                 jerseyResponse.setEntityStream(new InputStream() {
                     @Override
@@ -110,44 +120,29 @@
                     }
                 });
             }
-
-            if (asyncConnectorCallback != null) {
-                connector.executorService.execute(new Runnable() {
-                    @Override
-                    public void run() {
-                        asyncConnectorCallback.response(jerseyResponse);
-                        future.complete(jerseyResponse);
-                    }
-                });
-            }
-
         }
         if (msg instanceof HttpContent) {
+
             HttpContent httpContent = (HttpContent) msg;
 
             ByteBuf content = httpContent.content();
+
             if (content.isReadable()) {
                 content.retain();
-                isList.add(content);
+                nis.publish(content);
             }
 
             if (msg instanceof LastHttpContent) {
-                isList.add(Unpooled.EMPTY_BUFFER);
+                responseDone.complete(null);
+                notifyResponse();
             }
         }
     }
 
+
+
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, final Throwable cause) {
-        if (asyncConnectorCallback != null) {
-            connector.executorService.execute(new Runnable() {
-                @Override
-                public void run() {
-                    asyncConnectorCallback.failure(cause);
-                }
-            });
-        }
-        future.completeExceptionally(cause);
-        ctx.close();
+        responseDone.completeExceptionally(cause);
     }
 }
diff --git a/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/NettyConnector.java b/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/NettyConnector.java
index 2b36006..373d3c8 100644
--- a/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/NettyConnector.java
+++ b/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/NettyConnector.java
@@ -20,6 +20,8 @@
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.net.URI;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -28,7 +30,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 
 import javax.ws.rs.ProcessingException;
 import javax.ws.rs.client.Client;
@@ -76,6 +77,7 @@
     final ExecutorService executorService;
     final EventLoopGroup group;
     final Client client;
+    final HashMap<String, ArrayList<Channel>> connections = new HashMap<>();
 
     NettyConnector(Client client) {
 
@@ -83,72 +85,72 @@
 
         if (threadPoolSize != null && threadPoolSize instanceof Integer && (Integer) threadPoolSize > 0) {
             executorService = Executors.newFixedThreadPool((Integer) threadPoolSize);
+            this.group = new NioEventLoopGroup((Integer) threadPoolSize);
         } else {
             executorService = Executors.newCachedThreadPool();
+            this.group = new NioEventLoopGroup();
         }
 
-        this.group = new NioEventLoopGroup();
         this.client = client;
     }
 
     @Override
     public ClientResponse apply(ClientRequest jerseyRequest) {
-
-        final AtomicReference<ClientResponse> syncResponse = new AtomicReference<>(null);
-        final AtomicReference<Throwable> syncException = new AtomicReference<>(null);
-
         try {
-            Future<?> resultFuture = apply(jerseyRequest, new AsyncConnectorCallback() {
-                @Override
-                public void response(ClientResponse response) {
-                    syncResponse.set(response);
-                }
-
-                @Override
-                public void failure(Throwable failure) {
-                    syncException.set(failure);
-                }
-            });
+            CompletableFuture<ClientResponse> resultFuture = execute(jerseyRequest);
 
             Integer timeout = ClientProperties.getValue(jerseyRequest.getConfiguration().getProperties(),
                                                         ClientProperties.READ_TIMEOUT, 0);
 
-            if (timeout != null && timeout > 0) {
-                resultFuture.get(timeout, TimeUnit.MILLISECONDS);
-            } else {
-                resultFuture.get();
-            }
+            return (timeout != null && timeout > 0) ? resultFuture.get(timeout, TimeUnit.MILLISECONDS)
+                                                    : resultFuture.get();
         } catch (ExecutionException ex) {
             Throwable e = ex.getCause() == null ? ex : ex.getCause();
             throw new ProcessingException(e.getMessage(), e);
         } catch (Exception ex) {
             throw new ProcessingException(ex.getMessage(), ex);
         }
-
-        Throwable throwable = syncException.get();
-        if (throwable == null) {
-            return syncResponse.get();
-        } else {
-            throw new RuntimeException(throwable);
-        }
     }
 
     @Override
     public Future<?> apply(final ClientRequest jerseyRequest, final AsyncConnectorCallback jerseyCallback) {
+        return execute(jerseyRequest).whenCompleteAsync((r, th) -> {
+                  if (th == null) jerseyCallback.response(r);
+                  else jerseyCallback.failure(th);
+               }, executorService);
+    }
 
-        final CompletableFuture<Object> settableFuture = new CompletableFuture<>();
+    protected CompletableFuture<ClientResponse> execute(final ClientRequest jerseyRequest) {
+        final CompletableFuture<ClientResponse> responseAvailable = new CompletableFuture<>();
+        final CompletableFuture<?> responseDone = new CompletableFuture<>();
 
         final URI requestUri = jerseyRequest.getUri();
         String host = requestUri.getHost();
         int port = requestUri.getPort() != -1 ? requestUri.getPort() : "https".equals(requestUri.getScheme()) ? 443 : 80;
 
         try {
-            Bootstrap b = new Bootstrap();
-            b.group(group)
-             .channel(NioSocketChannel.class)
-             .handler(new ChannelInitializer<SocketChannel>() {
-                 @Override
-                 protected void initChannel(SocketChannel ch) throws Exception {
+            String key = requestUri.getScheme() + "://" + host + ":" + port;
+            ArrayList<Channel> conns;
+            synchronized (connections) {
+               conns = connections.get(key);
+               if (conns == null) {
+                  conns = new ArrayList<>(0);
+                  connections.put(key, conns);
+               }
+            }
+
+            Channel chan;
+            synchronized (conns) {
+               chan = conns.size() == 0 ? null : conns.remove(conns.size() - 1);
+            }
+
+            if (chan == null) {
+               Bootstrap b = new Bootstrap();
+               b.group(group)
+                .channel(NioSocketChannel.class)
+                .handler(new ChannelInitializer<SocketChannel>() {
+                    @Override
+                    protected void initChannel(SocketChannel ch) throws Exception {
                      ChannelPipeline p = ch.pipeline();
 
                      // Enable HTTPS if necessary.
@@ -177,32 +179,47 @@
                      p.addLast(new HttpClientCodec());
                      p.addLast(new ChunkedWriteHandler());
                      p.addLast(new HttpContentDecompressor());
-                     p.addLast(new JerseyClientHandler(NettyConnector.this, jerseyRequest, jerseyCallback, settableFuture));
-                 }
-             });
+                    }
+                });
 
-            // connect timeout
-            Integer connectTimeout = ClientProperties.getValue(jerseyRequest.getConfiguration().getProperties(),
-                                                               ClientProperties.CONNECT_TIMEOUT, 0);
-            if (connectTimeout > 0) {
-                b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout);
+               // connect timeout
+               Integer connectTimeout = ClientProperties.getValue(jerseyRequest.getConfiguration().getProperties(),
+                                                                  ClientProperties.CONNECT_TIMEOUT, 0);
+               if (connectTimeout > 0) {
+                   b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout);
+               }
+
+               // Make the connection attempt.
+               chan = b.connect(host, port).sync().channel();
             }
 
-            // Make the connection attempt.
-            final Channel ch = b.connect(host, port).sync().channel();
+            // assert: clientHandler will always notify responseDone: either normally, or exceptionally
+            // assert: clientHandler may notify responseAvailable, if sufficient parts of response are detected to construct
+            //         a valid ClientResponse
+            // assert: responseAvailable completion may be racing against responseDone completion
+            // assert: it is ok to abort the entire response, if responseDone is completed exceptionally - in particular, nothing
+            //         will leak
+            final Channel ch = chan;
+            JerseyClientHandler clientHandler = new JerseyClientHandler(jerseyRequest, responseAvailable, responseDone);
+            ch.pipeline().addLast(clientHandler);
 
-            // guard against prematurely closed channel
-            final GenericFutureListener<io.netty.util.concurrent.Future<? super Void>> closeListener =
-                    new GenericFutureListener<io.netty.util.concurrent.Future<? super Void>>() {
-                        @Override
-                        public void operationComplete(io.netty.util.concurrent.Future<? super Void> future) throws Exception {
-                            if (!settableFuture.isDone()) {
-                                settableFuture.completeExceptionally(new IOException("Channel closed."));
-                            }
-                        }
-                    };
+            responseDone.whenComplete((_r, th) -> {
+               ch.pipeline().remove(clientHandler);
 
-            ch.closeFuture().addListener(closeListener);
+               if (th == null) {
+                  synchronized (connections) {
+                     ArrayList<Channel> conns1 = connections.get(key);
+                     synchronized (conns1) {
+                        conns1.add(ch);
+                     }
+                  }
+               } else {
+                  ch.close();
+                  // if responseAvailable has been completed, no-op: jersey will encounter IOException while reading response body
+                  // if responseAvailable has not been completed, abort
+                  responseAvailable.completeExceptionally(th);
+               }
+            });
 
             HttpRequest nettyRequest;
             String pathWithQuery = buildPathWithQueryParameters(requestUri);
@@ -226,14 +243,23 @@
             nettyRequest.headers().add(HttpHeaderNames.HOST, jerseyRequest.getUri().getHost());
 
             if (jerseyRequest.hasEntity()) {
+                // guard against prematurely closed channel
+                final GenericFutureListener<io.netty.util.concurrent.Future<? super Void>> closeListener =
+                    new GenericFutureListener<io.netty.util.concurrent.Future<? super Void>>() {
+                        @Override
+                        public void operationComplete(io.netty.util.concurrent.Future<? super Void> future) throws Exception {
+                            if (!responseDone.isDone()) {
+                                responseDone.completeExceptionally(new IOException("Channel closed."));
+                            }
+                        }
+                    };
+                ch.closeFuture().addListener(closeListener);
                 if (jerseyRequest.getLengthLong() == -1) {
                     HttpUtil.setTransferEncodingChunked(nettyRequest, true);
                 } else {
                     nettyRequest.headers().add(HttpHeaderNames.CONTENT_LENGTH, jerseyRequest.getLengthLong());
                 }
-            }
 
-            if (jerseyRequest.hasEntity()) {
                 // Send the HTTP request.
                 ch.writeAndFlush(nettyRequest);
 
@@ -260,27 +286,22 @@
                         try {
                             jerseyRequest.writeEntity();
                         } catch (IOException e) {
-                            jerseyCallback.failure(e);
-                            settableFuture.completeExceptionally(e);
+                            responseDone.completeExceptionally(e);
                         }
                     }
                 });
 
                 ch.flush();
             } else {
-                // close listener is not needed any more.
-                ch.closeFuture().removeListener(closeListener);
-
                 // Send the HTTP request.
                 ch.writeAndFlush(nettyRequest);
             }
 
         } catch (InterruptedException e) {
-            settableFuture.completeExceptionally(e);
-            return settableFuture;
+            responseDone.completeExceptionally(e);
         }
 
-        return settableFuture;
+        return responseAvailable;
     }
 
     private String buildPathWithQueryParameters(URI requestUri) {
diff --git a/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/internal/NettyInputStream.java b/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/internal/NettyInputStream.java
index 741121f..3da7019 100644
--- a/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/internal/NettyInputStream.java
+++ b/connectors/netty-connector/src/main/java/org/glassfish/jersey/netty/connector/internal/NettyInputStream.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -16,107 +16,150 @@
 
 package org.glassfish.jersey.netty.connector.internal;
 
-import java.io.IOException;
 import java.io.InputStream;
-import java.util.concurrent.LinkedBlockingDeque;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
 
 /**
  * Input stream which servers as Request entity input.
  * <p>
- * Consumes a list of pending {@link ByteBuf}s and processes them on request by Jersey
+ * Converts Netty NIO buffers to an input streams and stores them in the queue,
+ * waiting for Jersey to process it.
+ *
+ * @author Pavel Bucek
  */
 public class NettyInputStream extends InputStream {
 
-    private final LinkedBlockingDeque<ByteBuf> isList;
+    private volatile boolean end = false;
+    private Throwable cause;
 
-    public NettyInputStream(LinkedBlockingDeque<ByteBuf> isList) {
-        this.isList = isList;
+    private final ArrayDeque<ByteBuf> isList;
+    private ByteBuf current;
+    private ByteBuffer buffer;
+
+    private byte[] ONE_BYTE;
+    private boolean reading;
+
+    public NettyInputStream() {
+        this.isList = new ArrayDeque<>();
     }
 
     @Override
     public int read(byte[] b, int off, int len) throws IOException {
-
-        ByteBuf take;
-        try {
-            take = isList.take();
-            boolean isReadable = take.isReadable();
-            int read = -1;
-            if (checkEndOfInputOrError(take)) {
-                take.release();
+       if (current == null) {
+          buffer = awaitNext();
+          if (buffer == null) {
+             // assert: end is true
+             if (cause == null) {
                 return -1;
-            }
+             }
 
-            if (isReadable) {
-                int readableBytes = take.readableBytes();
-                read = Math.min(readableBytes, len);
-                take.readBytes(b, off, read);
-                if (read < len) {
-                    take.release();
-                } else {
-                    isList.addFirst(take);
-                }
-            } else {
-                read = 0;
-                take.release(); //We don't need `0`
-            }
+             throw new IOException(cause);
+          }
+       }
 
-            return read;
-        } catch (InterruptedException e) {
-            throw new IOException("Interrupted.", e);
-        }
+       int rem = buffer.remaining();
+       if (rem < len) {
+          len = rem;
+       }
+       buffer.get(b, off, len);
+       if (rem == len) {
+          releaseByteBuf();
+       }
+
+       return len;
     }
 
     @Override
     public int read() throws IOException {
+       if (ONE_BYTE == null) {
+          ONE_BYTE = new byte[1];
+       }
+       int r = read(ONE_BYTE, 0, 1);
+       if (r < 0) {
+          return r;
+       }
 
-        ByteBuf take;
-        try {
-            take = isList.take();
-            boolean isReadable = take.isReadable();
-            if (checkEndOfInputOrError(take)) {
-                take.release();
-                return -1;
-            }
-
-            if (isReadable) {
-                return take.readInt();
-            } else {
-                take.release(); //We don't need `0`
-            }
-
-            return 0;
-        } catch (InterruptedException e) {
-            throw new IOException("Interrupted.", e);
-        }
+       return ONE_BYTE[0] & 0xff;
     }
 
     @Override
-    public void close() throws IOException {
-        if (isList != null) {
-            while (!isList.isEmpty()) {
-                try {
-                    isList.take().release();
-                } catch (InterruptedException e) {
-                    throw new IOException("Interrupted. Potential ByteBuf Leak.", e);
-                }
-            }
+    public void close() {
+
+        releaseByteBuf();
+
+        cleanup(true);
+    }
+
+    private void releaseByteBuf() {
+        if (current != null) {
+            current.release();
         }
-        super.close();
+
+        current = null;
+        buffer = null;
+    }
+
+    protected synchronized ByteBuffer awaitNext() {
+       while (isList.isEmpty()) {
+          if (end) {
+             return null;
+          }
+
+          try {
+             reading = true;
+             wait();
+             reading = false;
+          } catch (InterruptedException ie) {
+             // waiting uninterruptibly
+          }
+       }
+
+       current = isList.poll();
+       return current.nioBuffer().asReadOnlyBuffer();
+    }
+
+    public void complete(Throwable cause) {
+       this.cause = cause;
+       cleanup(cause != null);
+    }
+
+    protected synchronized void cleanup(boolean drain) {
+       if (drain) {
+          while (!isList.isEmpty()) {
+             isList.poll().release();
+          }
+       }
+
+       end = true;
+
+       if (reading) {
+          notifyAll();
+       }
     }
 
     @Override
     public int available() throws IOException {
-        ByteBuf peek = isList.peek();
-        if (peek != null && peek.isReadable()) {
-            return peek.readableBytes();
-        }
-        return 0;
+        return buffer == null ? 0 : buffer.remaining();
     }
 
-    private boolean checkEndOfInputOrError(ByteBuf take) throws IOException {
-        return take == Unpooled.EMPTY_BUFFER;
+    public synchronized void publish(ByteBuf content) {
+       if (end || content.nioBuffer().remaining() == 0) {
+          content.release();
+          return;
+       }
+
+       isList.add(content);
+       if (reading) {
+          notifyAll();
+       }
+    }
+
+    public void clear() {
+        end = false;
+        isList.clear();
     }
 }
diff --git a/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/HelloWorldTest.java b/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/HelloWorldTest.java
index 6716c9b..20caa53 100644
--- a/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/HelloWorldTest.java
+++ b/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/HelloWorldTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -77,6 +77,7 @@
     public void testConnection() {
         Response response = target().path(ROOT_PATH).request("text/plain").get();
         assertEquals(200, response.getStatus());
+        response.close();
     }
 
     @Test
diff --git a/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/HttpHeadersTest.java b/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/HttpHeadersTest.java
index 10cfa14..f5438e5 100644
--- a/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/HttpHeadersTest.java
+++ b/connectors/netty-connector/src/test/java/org/glassfish/jersey/netty/connector/HttpHeadersTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2018 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -64,5 +64,6 @@
 
         assertEquals(200, response.getStatus());
         assertTrue(response.hasEntity());
+        response.close();
     }
 }
diff --git a/containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyHttp2ServerHandler.java b/containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyHttp2ServerHandler.java
index e64476d..6eeac17 100644
--- a/containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyHttp2ServerHandler.java
+++ b/containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyHttp2ServerHandler.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -25,11 +25,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.LinkedBlockingDeque;
 
 import javax.ws.rs.core.SecurityContext;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -55,7 +52,7 @@
 class JerseyHttp2ServerHandler extends ChannelDuplexHandler {
 
     private final URI baseUri;
-    private final LinkedBlockingDeque<ByteBuf> isList = new LinkedBlockingDeque<>();
+    private final NettyInputStream nettyInputStream = new NettyInputStream();
     private final NettyHttpContainer container;
     private final ResourceConfig resourceConfig;
 
@@ -92,9 +89,9 @@
      * Process incoming data.
      */
     private void onDataRead(ChannelHandlerContext ctx, Http2DataFrame data) throws Exception {
-        isList.add(data.content());
+        nettyInputStream.publish(data.content());
         if (data.isEndStream()) {
-            isList.add(Unpooled.EMPTY_BUFFER);
+            nettyInputStream.complete(null);
         }
     }
 
@@ -163,11 +160,11 @@
             ctx.channel().closeFuture().addListener(new GenericFutureListener<Future<? super Void>>() {
                 @Override
                 public void operationComplete(Future<? super Void> future) throws Exception {
-                    isList.add(Unpooled.EMPTY_BUFFER);
+                    nettyInputStream.complete(future.cause());
                 }
             });
 
-            requestContext.setEntityStream(new NettyInputStream(isList));
+            requestContext.setEntityStream(nettyInputStream);
         } else {
             requestContext.setEntityStream(new InputStream() {
                 @Override
diff --git a/containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyServerHandler.java b/containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyServerHandler.java
index 712cb1f..0f2a7ae 100644
--- a/containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyServerHandler.java
+++ b/containers/netty-http/src/main/java/org/glassfish/jersey/netty/httpserver/JerseyServerHandler.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2019 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020 Oracle and/or its affiliates. All rights reserved.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0, which is available at
@@ -20,13 +20,11 @@
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.LinkedBlockingDeque;
 
 import javax.ws.rs.core.Response.Status;
 import javax.ws.rs.core.MediaType;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.http.DefaultFullHttpResponse;
@@ -53,7 +51,7 @@
 class JerseyServerHandler extends ChannelInboundHandlerAdapter {
 
     private final URI baseUri;
-    private final LinkedBlockingDeque<ByteBuf> isList = new LinkedBlockingDeque<>();
+    private final NettyInputStream nettyInputStream = new NettyInputStream();
     private final NettyHttpContainer container;
     private final ResourceConfig resourceConfig;
 
@@ -82,7 +80,7 @@
                 ctx.write(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE));
             }
 
-            isList.clear(); // clearing the content - possible leftover from previous request processing.
+            nettyInputStream.clear(); // clearing the content - possible leftover from previous request processing.
             final ContainerRequest requestContext = createContainerRequest(ctx, req);
 
             requestContext.setWriter(new NettyResponseWriter(ctx, req, container));
@@ -105,7 +103,7 @@
                 //Otherwise, it's safe to discard during next processing
                 if ((!isJson && contentLength != -1) || HttpUtil.isTransferEncodingChunked(req)
                         || (isJson && contentLength >= 2)) {
-                    requestContext.setEntityStream(new NettyInputStream(isList));
+                    requestContext.setEntityStream(nettyInputStream);
                 }
             }
 
@@ -128,11 +126,11 @@
 
           ByteBuf content = httpContent.content();
           if (content.isReadable()) {
-              isList.add(content);
+              nettyInputStream.publish(content);
           }
 
           if (msg instanceof LastHttpContent) {
-              isList.add(Unpooled.EMPTY_BUFFER);
+              nettyInputStream.complete(null);
           }
       }
     }
diff --git a/pom.xml b/pom.xml
index a157ec4..3cad108 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2141,7 +2141,7 @@
         <mockito.version>1.10.19</mockito.version>
         <moxy.version>2.7.4</moxy.version>
         <mustache.version>0.8.17</mustache.version>
-        <netty.version>4.1.31.Final</netty.version>
+        <netty.version>4.1.43.Final</netty.version>
         <nexus-staging.mvn.plugin.version>1.6.7</nexus-staging.mvn.plugin.version>
         <opentracing.version>0.30.0</opentracing.version>
         <osgi.version>6.0.0</osgi.version>