deadlock fix attempt v4

sdk/clientcore/http-netty4/spotbugs-exclude.xml

@@ -62,4 +62,8 @@
     <Bug pattern="VO_VOLATILE_REFERENCE_TO_ARRAY" />
     <Class name="io.clientcore.http.netty4.implementation.Netty4ChannelBinaryData" />
   </Match>
+  <Match>
+    <Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE" />
+    <Class name="io.clientcore.http.netty4.implementation.Netty4ChannelInputStream" />
+  </Match>
 </FindBugsFilter>

sdk/clientcore/http-netty4/src/main/java/io/clientcore/http/netty4/implementation/Netty4ChannelInputStream.java

@@ -7,7 +7,8 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.LinkedList;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 
 /**
@@ -28,9 +29,9 @@ public final class Netty4ChannelInputStream extends InputStream {
     // Once this is true, the stream will never return data again.
     private boolean streamDone = false;
 
-    // Linked list of byte[]s that maintains the last available contents from the Channel / eager content.
-    // A list is needed as each Channel.read() may result in many channelRead calls.
-    private final LinkedList<byte[]> additionalBuffers;
+    // Queue of byte[]s that maintains the last available contents from the Channel / eager content.
+    // A queue is needed as each Channel.read() may result in many channelRead calls.
+    private final Queue<byte[]> additionalBuffers;
 
     private byte[] currentBuffer;
 
@@ -57,7 +58,7 @@ public final class Netty4ChannelInputStream extends InputStream {
             this.currentBuffer = new byte[0];
         }
         this.readIndex = 0;
-        this.additionalBuffers = new LinkedList<>();
+        this.additionalBuffers = new ConcurrentLinkedQueue<>();
         this.channel = channel;
         if (channel.pipeline().get(Netty4InitiateOneReadHandler.class) != null) {
             channel.pipeline().remove(Netty4InitiateOneReadHandler.class);
@@ -190,7 +191,7 @@ public void close() throws IOException {
 
     private boolean setupNextBuffer() throws IOException {
         if (!additionalBuffers.isEmpty()) {
-            currentBuffer = additionalBuffers.pop();
+            currentBuffer = additionalBuffers.poll();
             readIndex = 0;
             return true;
         } else if (readMore()) {
@@ -225,7 +226,7 @@ private boolean readMore() throws IOException {
                 byte[] buffer = new byte[byteBuf.readableBytes()];
                 byteBuf.readBytes(buffer);
 
-                additionalBuffers.add(buffer);
+                additionalBuffers.offer(buffer);
             }, isHttp2);
             channel.pipeline().addLast(Netty4HandlerNames.READ_ONE, handler);
         }
@@ -253,7 +254,7 @@ private boolean readMore() throws IOException {
         }
 
         if (!additionalBuffers.isEmpty()) {
-            currentBuffer = additionalBuffers.pop();
+            currentBuffer = additionalBuffers.poll();
             readIndex = 0;
         } else if (channelDone) { // Don't listen to IntelliJ here, channelDone may be false.
             // This read contained no data and the channel completed, therefore the stream is also completed.

sdk/clientcore/http-netty4/src/main/java/io/clientcore/http/netty4/implementation/Netty4EagerConsumeChannelHandler.java

@@ -62,23 +62,23 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
         } catch (IOException | RuntimeException ex) {
             ReferenceCountUtil.release(msg);
             ctx.fireExceptionCaught(ex);
-            cleanup(ctx);
+            latch.countDown();
         }
     }
 
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) {
         ctx.fireChannelReadComplete();
         if (lastRead) {
-            cleanup(ctx);
+            latch.countDown();
         }
     }
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         this.exception = cause;
         ctx.fireExceptionCaught(cause);
-        cleanup(ctx);
+        latch.countDown();
     }
 
     Throwable channelException() {
@@ -88,13 +88,13 @@ Throwable channelException() {
     // TODO (alzimmer): Are the latch countdowns needed for unregistering and inactivity?
     @Override
     public void channelUnregistered(ChannelHandlerContext ctx) {
-        cleanup(ctx);
+        latch.countDown();
         ctx.fireChannelUnregistered();
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) {
-        cleanup(ctx);
+        latch.countDown();
         ctx.fireChannelInactive();
     }
 
@@ -105,15 +105,6 @@ public void handlerAdded(ChannelHandlerContext ctx) {
             // an exception. Simply counting down the latch would cause the caller to receive
             // an empty/incomplete data stream without any indication of the underlying network error.
             ctx.fireExceptionCaught(new ClosedChannelException());
-            ctx.pipeline().remove(this);
-        }
-    }
-
-    private void cleanup(ChannelHandlerContext ctx) {
-        if (ctx.pipeline().get(Netty4EagerConsumeChannelHandler.class) != null) {
-            ctx.pipeline().remove(this);
         }
-
-        latch.countDown();
     }
 }

sdk/clientcore/http-netty4/src/main/java/io/clientcore/http/netty4/implementation/Netty4InitiateOneReadHandler.java

@@ -95,9 +95,6 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) {
         latch.countDown();
-        if (lastRead) {
-            ctx.pipeline().remove(this);
-        }
         ctx.fireChannelReadComplete();
     }
 
@@ -109,9 +106,6 @@ boolean isChannelConsumed() {
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         this.exception = cause;
         latch.countDown();
-        if (ctx.pipeline().get(Netty4InitiateOneReadHandler.class) != null) {
-            ctx.pipeline().remove(this);
-        }
         ctx.fireExceptionCaught(cause);
     }
 
@@ -123,18 +117,12 @@ Throwable channelException() {
     @Override
     public void channelUnregistered(ChannelHandlerContext ctx) {
         latch.countDown();
-        if (ctx.pipeline().get(Netty4InitiateOneReadHandler.class) != null) {
-            ctx.pipeline().remove(this);
-        }
         ctx.fireChannelUnregistered();
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) {
         latch.countDown();
-        if (ctx.pipeline().get(Netty4InitiateOneReadHandler.class) != null) {
-            ctx.pipeline().remove(this);
-        }
         ctx.fireChannelInactive();
     }
 
@@ -145,7 +133,6 @@ public void handlerAdded(ChannelHandlerContext ctx) {
             // an exception. Simply counting down the latch would cause the caller to receive
             // an empty/incomplete data stream without any indication of the underlying network error.
             ctx.fireExceptionCaught(new ClosedChannelException());
-            ctx.pipeline().remove(this);
         }
     }
 }

sdk/clientcore/http-netty4/src/main/java/io/clientcore/http/netty4/implementation/Netty4PipelineCleanupHandler.java

@@ -6,8 +6,6 @@
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPipeline;
-import io.netty.handler.codec.http.LastHttpContent;
-import io.netty.handler.codec.http2.Http2DataFrame;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -30,7 +28,6 @@ public class Netty4PipelineCleanupHandler extends ChannelDuplexHandler {
 
     private final Netty4ConnectionPool connectionPool;
     private final AtomicBoolean cleanedUp = new AtomicBoolean(false);
-    private boolean lastContentRead;
 
     private static final List<String> HANDLERS_TO_REMOVE;
 
@@ -50,46 +47,29 @@ public Netty4PipelineCleanupHandler(Netty4ConnectionPool connectionPool) {
         this.connectionPool = connectionPool;
     }
 
-    @Override
-    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        if (msg instanceof LastHttpContent) {
-            this.lastContentRead = true;
-        } else if (msg instanceof Http2DataFrame) {
-            this.lastContentRead = ((Http2DataFrame) msg).isEndStream();
-        }
-
-        ctx.fireChannelRead(msg);
-    }
-
-    @Override
-    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
-        // First, let other handlers process the channelReadComplete event.
-        ctx.fireChannelReadComplete();
-
-        if (lastContentRead) {
-            cleanup(ctx);
-        }
-    }
-
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         // An exception has occurred, which means the channel is likely in a bad state.
         // We handle this by closing the channel. This prevents it from being
         // returned to the connection pool.
-        ctx.close();
+        cleanup(ctx, true);
     }
 
     @Override
-    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+    public void channelInactive(ChannelHandlerContext ctx) {
         ctx.fireChannelInactive();
-        cleanup(ctx);
+        cleanup(ctx, true);
     }
 
-    public void cleanup(ChannelHandlerContext ctx) {
+    public void cleanup(ChannelHandlerContext ctx, boolean closeChannel) {
         if (!cleanedUp.compareAndSet(false, true)) {
             return;
         }
 
+        // Always reset autoRead to false before returning a channel to the pool
+        // to ensure predictable behavior for the next request.
+        ctx.channel().config().setAutoRead(false);
+
         ChannelPipeline pipeline = ctx.channel().pipeline();
         for (String handlerName : HANDLERS_TO_REMOVE) {
             if (pipeline.get(handlerName) != null) {
@@ -101,6 +81,10 @@ public void cleanup(ChannelHandlerContext ctx) {
             pipeline.remove(this);
         }
 
-        connectionPool.release(ctx.channel());
+        if (closeChannel) {
+            ctx.channel().close();
+        } else {
+            connectionPool.release(ctx.channel());
+        }
     }
 }