Release pipelined http responses on close

[Tim-Brooks]

Right now it is possible for the HttpPipeliningHandler to queue
pipelined responses. On channel close, we do not clear and release these
responses. This commit releases the responses and completes the promises.

[Tim-Brooks]

@jasontedor Is there something that I am missing here and this is not necessary? I do not see anywhere else where these would be released.

[jasontedor]

Very good catch @tbrooks8, I left one piece of feedback.

[jasontedor]

Only use one ClosedChannelException?

[jasontedor]

Maybe guard this and the loop with a check that the holding queue is not empty?

[Tim-Brooks]

I'm not exactly sure what you mean. Do you mean something like this:

synchronized (holdingQueue) {
            for (HttpPipelinedResponse pipelinedResponse : holdingQueue) {
                pipelinedResponse.release();
                pipelinedResponse.promise().setFailure(closedChannelException);
            }
            holdingQueue.clear();
        }

[Tim-Brooks]

Or, I guess you mean this?

        synchronized (holdingQueue) {
            while (holdingQueue.isEmpty() == false) {
                HttpPipelinedResponse pipelinedResponse = holdingQueue.poll();
                pipelinedResponse.release();
                pipelinedResponse.promise().setFailure(closedChannelException);
            }
        }

[jasontedor]

I don't think the synchronization is necessary (I think it can actually all be removed from this class) but let's put that aside for now as a separate issue.

I meant:

diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/pipelining/HttpPipeliningHandler.java b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/pipelining/HttpPipeliningHandler.java
index c09099a221..e9d99378e2 100644
--- a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/pipelining/HttpPipeliningHandler.java
+++ b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/pipelining/HttpPipeliningHandler.java
@@ -123,10 +123,12 @@ public class HttpPipeliningHandler extends ChannelDuplexHandler {
 
     @Override
     public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
-        ClosedChannelException closedChannelException = new ClosedChannelException();
-        for (HttpPipelinedResponse pipelinedResponse : holdingQueue) {
-            pipelinedResponse.release();
-            pipelinedResponse.promise().setFailure(closedChannelException);
+        if (!holdingQueue.isEmpty()) {
+            final ClosedChannelException closedChannelException = new ClosedChannelException();
+            for (final HttpPipelinedResponse pipelinedResponse : holdingQueue) {
+                pipelinedResponse.release();
+                pipelinedResponse.promise().setFailure(closedChannelException);
+            }
         }
         ctx.close(promise);
     }

(and sure, we can add the synchronized block for now until I get around to investigating removing it).

[Tim-Brooks]

Okay. I will make a change based on that. And I agree that synchronization is not necessary (as it should probably be unnecessary in the write method). I just was confused by what you meant by guard.

[jasontedor]

Yeah that's what I mean by removing it completely from this class. Sorry for the confusion.

[s1monw]

left one question LGTM though

[s1monw]

should we catch here and continue in the case of a failure?

[jasontedor]

@tbrooks8 I updated the labels, I think this should be taken to 5.6 too.

[Tim-Brooks]

Merged and cherry-picked onto 6.x, 6.0, and 5.6