JERSEY-2558: ChunkedOutput.close() broken since Jersey 2.8

 - Fix in ServerRuntime and AsyncContextDelegateProviderImpl
   - Problem seems to be only in servlet container when ChunkedOutput is closed before
     response from resource method is processed by Jersey - request is indefinitely
     suspended after it's completed (and never completed after that).

 - Added test for 2558 and also for combination of ChunkedOutput with AsyncResponse

Change-Id: I705f139118d6a1a236a28aac0f78e02847d9b24a
Signed-off-by: Michal Gajdos <[email protected]>

Author: Michal Gajdos

containers/jersey-servlet-core/src/main/java/org/glassfish/jersey/servlet/internal/ResponseWriter.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2012-2014 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012-2015 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -119,7 +119,7 @@ public boolean suspend(final long timeOut, final TimeUnit timeUnit, final Timeou
         try {
             // Suspend the servlet.
             asyncExt.suspend();
-        } catch (IllegalStateException ex) {
+        } catch (final IllegalStateException ex) {
             LOGGER.log(Level.WARNING, LocalizationMessages.SERVLET_REQUEST_SUSPEND_FAILED(), ex);
             return false;
         }
@@ -128,12 +128,12 @@ public boolean suspend(final long timeOut, final TimeUnit timeUnit, final Timeou
     }
 
     @Override
-    public void setSuspendTimeout(long timeOut, TimeUnit timeUnit) throws IllegalStateException {
+    public void setSuspendTimeout(final long timeOut, final TimeUnit timeUnit) throws IllegalStateException {
         requestTimeoutHandler.setSuspendTimeout(timeOut, timeUnit);
     }
 
     @Override
-    public OutputStream writeResponseStatusAndHeaders(long contentLength, ContainerResponse responseContext)
+    public OutputStream writeResponseStatusAndHeaders(final long contentLength, final ContainerResponse responseContext)
             throws ContainerException {
         this.responseContext.set(responseContext);
 
@@ -145,8 +145,8 @@ public OutputStream writeResponseStatusAndHeaders(long contentLength, ContainerR
         // the invocation of sendError as on some Servlet implementations
         // modification of the response headers will have no effect
         // after the invocation of sendError.
-        MultivaluedMap<String, String> headers = getResponseContext().getStringHeaders();
-        for (Map.Entry<String, List<String>> e : headers.entrySet()) {
+        final MultivaluedMap<String, String> headers = getResponseContext().getStringHeaders();
+        for (final Map.Entry<String, List<String>> e : headers.entrySet()) {
             final Iterator<String> it = e.getValue().iterator();
             if (!it.hasNext()) {
                 continue;
@@ -178,7 +178,7 @@ public OutputStream writeResponseStatusAndHeaders(long contentLength, ContainerR
                 // delegating output stream prevents closing the underlying servlet output stream,
                 // so that any Servlet filters in the chain can still write to the response after us.
                 return new NonCloseableOutputStreamWrapper(outputStream);
-            } catch (IOException e) {
+            } catch (final IOException e) {
                 throw new ContainerException(e);
             }
         }
@@ -198,7 +198,7 @@ public void commit() {
                         } else {
                             response.sendError(status, reason);
                         }
-                    } catch (IOException ex) {
+                    } catch (final IOException ex) {
                         throw new ContainerException(
                                 LocalizationMessages.EXCEPTION_SENDING_ERROR_RESPONSE(status, reason != null ? reason : "--"),
                                 ex);
@@ -212,7 +212,7 @@ public void commit() {
     }
 
     @Override
-    public void failure(Throwable error) {
+    public void failure(final Throwable error) {
         try {
             if (!response.isCommitted()) {
                 try {
@@ -223,10 +223,10 @@ public void failure(Throwable error) {
                     } else {
                         response.sendError(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), "Request failed.");
                     }
-                } catch (IllegalStateException ex) {
+                } catch (final IllegalStateException ex) {
                     // a race condition externally committing the response can still occur...
                     LOGGER.log(Level.FINER, "Unable to reset failed response.", ex);
-                } catch (IOException ex) {
+                } catch (final IOException ex) {
                     throw new ContainerException(LocalizationMessages.EXCEPTION_SENDING_ERROR_RESPONSE(
                             Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), "Request failed."), ex);
                 } finally {
@@ -249,7 +249,7 @@ public boolean enableResponseBuffering() {
      *
      * @param error throwable to be re-thrown
      */
-    private void rethrow(Throwable error) {
+    private void rethrow(final Throwable error) {
         if (error instanceof RuntimeException) {
             throw (RuntimeException) error;
         } else {
@@ -279,22 +279,22 @@ private ContainerResponse getResponseContext() {
     private static class NonCloseableOutputStreamWrapper extends OutputStream {
         private final OutputStream delegate;
 
-        public NonCloseableOutputStreamWrapper(OutputStream delegate) {
+        public NonCloseableOutputStreamWrapper(final OutputStream delegate) {
             this.delegate = delegate;
         }
 
         @Override
-        public void write(int b) throws IOException {
+        public void write(final int b) throws IOException {
             delegate.write(b);
         }
 
         @Override
-        public void write(byte[] b) throws IOException {
+        public void write(final byte[] b) throws IOException {
             delegate.write(b);
         }
 
         @Override
-        public void write(byte[] b, int off, int len) throws IOException {
+        public void write(final byte[] b, final int off, final int len) throws IOException {
             delegate.write(b, off, len);
         }
 

containers/jersey-servlet/src/main/java/org/glassfish/jersey/servlet/async/AsyncContextDelegateProviderImpl.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2012-2013 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012-2015 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -39,6 +39,7 @@
  */
 package org.glassfish.jersey.servlet.async;
 
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import javax.servlet.AsyncContext;
@@ -69,6 +70,7 @@ private static final class ExtensionImpl implements AsyncContextDelegate {
         private final HttpServletRequest request;
         private final HttpServletResponse response;
         private final AtomicReference<AsyncContext> asyncContextRef;
+        private final AtomicBoolean completed;
 
         /**
          * Create a Servlet 3.x {@link AsyncContextDelegate} with given {@code request} and {@code response}.
@@ -79,21 +81,27 @@ private static final class ExtensionImpl implements AsyncContextDelegate {
         private ExtensionImpl(final HttpServletRequest request, final HttpServletResponse response) {
             this.request = request;
             this.response = response;
-            this.asyncContextRef = new AtomicReference<AsyncContext>();
+            this.asyncContextRef = new AtomicReference<>();
+            this.completed = new AtomicBoolean(false);
         }
 
         @Override
         public void suspend() throws IllegalStateException {
-            final AsyncContext asyncContext = request.startAsync(request, response);
+            // Suspend only if not completed and not suspended before.
+            if (!completed.get() && asyncContextRef.get() == null) {
+                final AsyncContext asyncContext = request.startAsync(request, response);
 
-            // Tell underlying asyncContext to never time out.
-            asyncContext.setTimeout(NEVER_TIMEOUT_VALUE);
+                // Tell underlying asyncContext to never time out.
+                asyncContext.setTimeout(NEVER_TIMEOUT_VALUE);
 
-            asyncContextRef.set(asyncContext);
+                asyncContextRef.set(asyncContext);
+            }
         }
 
         @Override
         public void complete() {
+            completed.set(true);
+
             final AsyncContext asyncContext = asyncContextRef.getAndSet(null);
             if (asyncContext != null) {
                 asyncContext.complete();

core-client/src/main/java/org/glassfish/jersey/client/ChunkedInput.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2012-2013 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012-2015 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -70,7 +70,9 @@
  * @param <T> chunk type.
  * @author Marek Potociar (marek.potociar at oracle.com)
  */
+@SuppressWarnings("UnusedDeclaration")
 public class ChunkedInput<T> extends GenericType<T> implements Closeable {
+
     private static final Logger LOGGER = Logger.getLogger(ChunkedInput.class.getName());
 
     private final AtomicBoolean closed = new AtomicBoolean(false);
@@ -108,12 +110,12 @@ public static ChunkParser createParser(final byte[] boundary) {
     private static class FixedBoundaryParser implements ChunkParser {
         private final byte[] delimiter;
 
-        public FixedBoundaryParser(byte[] boundary) {
+        public FixedBoundaryParser(final byte[] boundary) {
             delimiter = Arrays.copyOf(boundary, boundary.length);
         }
 
         @Override
-        public byte[] readChunk(InputStream in) throws IOException {
+        public byte[] readChunk(final InputStream in) throws IOException {
             final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
             final byte[] delimiterBuffer = new byte[delimiter.length];
 
@@ -157,13 +159,13 @@ public byte[] readChunk(InputStream in) throws IOException {
      * @param propertiesDelegate properties delegate for this request/response.
      */
     protected ChunkedInput(
-            Type chunkType,
-            InputStream inputStream,
-            Annotation[] annotations,
-            MediaType mediaType,
-            MultivaluedMap<String, String> headers,
-            MessageBodyWorkers messageBodyWorkers,
-            PropertiesDelegate propertiesDelegate) {
+            final Type chunkType,
+            final InputStream inputStream,
+            final Annotation[] annotations,
+            final MediaType mediaType,
+            final MultivaluedMap<String, String> headers,
+            final MessageBodyWorkers messageBodyWorkers,
+            final PropertiesDelegate propertiesDelegate) {
         super(chunkType);
 
         this.inputStream = inputStream;
@@ -196,7 +198,7 @@ public ChunkParser getParser() {
      *
      * @param parser new chunk parser.
      */
-    public void setParser(ChunkParser parser) {
+    public void setParser(final ChunkParser parser) {
         this.parser = parser;
     }
 
@@ -236,7 +238,7 @@ public MediaType getChunkType() {
      * @param mediaType custom chunk data media type. Must not be {@code null}.
      * @throws IllegalArgumentException in case the {@code mediaType} is {@code null}.
      */
-    public void setChunkType(MediaType mediaType) throws IllegalArgumentException {
+    public void setChunkType(final MediaType mediaType) throws IllegalArgumentException {
         if (mediaType == null) {
             throw new IllegalArgumentException(LocalizationMessages.CHUNKED_INPUT_MEDIA_TYPE_NULL());
         }
@@ -255,7 +257,7 @@ public void setChunkType(MediaType mediaType) throws IllegalArgumentException {
      *                                  a valid {@link MediaType} instance or is {@code null}.
      * @see #setChunkType(javax.ws.rs.core.MediaType)
      */
-    public void setChunkType(String mediaType) throws IllegalArgumentException {
+    public void setChunkType(final String mediaType) throws IllegalArgumentException {
         this.mediaType = MediaType.valueOf(mediaType);
     }
 
@@ -265,7 +267,7 @@ public void close() {
             if (inputStream != null) {
                 try {
                     inputStream.close();
-                } catch (IOException e) {
+                } catch (final IOException e) {
                     LOGGER.log(Level.FINE, LocalizationMessages.CHUNKED_INPUT_STREAM_CLOSING_ERROR(), e);
                 }
             }
@@ -306,7 +308,7 @@ public T read() throws IllegalStateException {
             if (chunk == null) {
                 close();
             } else {
-                ByteArrayInputStream chunkStream = new ByteArrayInputStream(chunk);
+                final ByteArrayInputStream chunkStream = new ByteArrayInputStream(chunk);
                 // TODO: add interceptors: interceptors are used in ChunkedOutput, so the stream should
                 // be intercepted in the ChunkedInput too. Interceptors cannot be easily added to the readFrom
                 // method as they should wrap the stream before it is processed by ChunkParser. Also please check todo
@@ -323,7 +325,7 @@ public T read() throws IllegalStateException {
                         Collections.<ReaderInterceptor>emptyList(),
                         false);
             }
-        } catch (IOException e) {
+        } catch (final IOException e) {
             Logger.getLogger(this.getClass().getName()).log(Level.FINE, e.getMessage(), e);
             close();
         }

core-server/src/main/java/org/glassfish/jersey/server/ChunkedOutput.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2012-2014 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012-2015 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -106,7 +106,7 @@ public ChunkedOutput(final Type chunkType) {
      * @param chunkDelimiter custom chunk delimiter bytes. Must not be {code null}.
      * @since 2.4.1
      */
-    protected ChunkedOutput(byte[] chunkDelimiter) {
+    protected ChunkedOutput(final byte[] chunkDelimiter) {
         if (chunkDelimiter.length > 0) {
             this.chunkDelimiter = new byte[chunkDelimiter.length];
             System.arraycopy(chunkDelimiter, 0, this.chunkDelimiter, 0, chunkDelimiter.length);
@@ -122,7 +122,7 @@ protected ChunkedOutput(byte[] chunkDelimiter) {
      * @param chunkDelimiter custom chunk delimiter bytes. Must not be {code null}.
      * @since 2.4.1
      */
-    public ChunkedOutput(final Type chunkType, byte[] chunkDelimiter) {
+    public ChunkedOutput(final Type chunkType, final byte[] chunkDelimiter) {
         super(chunkType);
         if (chunkDelimiter.length > 0) {
             this.chunkDelimiter = new byte[chunkDelimiter.length];
@@ -138,7 +138,7 @@ public ChunkedOutput(final Type chunkType, byte[] chunkDelimiter) {
      * @param chunkDelimiter custom chunk delimiter string. Must not be {code null}.
      * @since 2.4.1
      */
-    protected ChunkedOutput(String chunkDelimiter) {
+    protected ChunkedOutput(final String chunkDelimiter) {
         if (chunkDelimiter.isEmpty()) {
             this.chunkDelimiter = ZERO_LENGTH_DELIMITER;
         } else {
@@ -153,7 +153,7 @@ protected ChunkedOutput(String chunkDelimiter) {
      * @param chunkDelimiter custom chunk delimiter string. Must not be {code null}.
      * @since 2.4.1
      */
-    public ChunkedOutput(final Type chunkType, String chunkDelimiter) {
+    public ChunkedOutput(final Type chunkType, final String chunkDelimiter) {
         super(chunkType);
         if (chunkDelimiter.isEmpty()) {
             this.chunkDelimiter = ZERO_LENGTH_DELIMITER;
@@ -243,10 +243,10 @@ public Void call() throws IOException {
                                 // if MBW replaced the stream, let's make sure to set it in the response context.
                                 responseContext.setEntityStream(writtenStream);
                             }
-                        } catch (IOException ioe) {
+                        } catch (final IOException ioe) {
                             connectionCallback.onDisconnect(asyncContext.get());
                             throw ioe;
-                        } catch (MappableException mpe) {
+                        } catch (final MappableException mpe) {
                             if (mpe.getCause() instanceof IOException) {
                                 connectionCallback.onDisconnect(asyncContext.get());
                             }
@@ -278,15 +278,15 @@ public Void call() throws IOException {
                     return null;
                 }
             });
-        } catch (Exception e) {
+        } catch (final Exception e) {
             closed = true;
             // remember the exception (it will get rethrown from finally clause, once it does it's work)
             ex = e;
         } finally {
             if (closed) {
                 try {
                     responseContext.close();
-                } catch (Exception e) {
+                } catch (final Exception e) {
                     // if no exception remembered before, remember this one
                     // otherwise the previously remembered exception (from catch clause) takes precedence
                     ex = ex == null ? e : ex;

core-server/src/main/java/org/glassfish/jersey/server/ServerRuntime.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2012-2014 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012-2015 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -665,8 +665,9 @@ public OutputStream getOutputStream(final int contentLength) throws IOException
                             close = true;
                         }
 
+                        final ChunkedOutput chunked = (ChunkedOutput) entity;
                         try {
-                            ((ChunkedOutput) entity).setContext(
+                            chunked.setContext(
                                     runtime.requestScope,
                                     runtime.requestScope.referenceCurrent(),
                                     request,
@@ -677,9 +678,11 @@ public OutputStream getOutputStream(final int contentLength) throws IOException
                             LOGGER.log(Level.SEVERE, LocalizationMessages.ERROR_WRITING_RESPONSE_ENTITY_CHUNK(), ex);
                             close = true;
                         }
-                        // suspend the writer indefinitely (passing null timeout handler is ok in such case).
+                        // suspend the writer indefinitely (passing null timeout handler is ok in such case) if the output is not
+                        // already closed.
                         // TODO what to do if we detect that the writer has already been suspended? override the timeout value?
-                        if (!writer.suspend(0, TimeUnit.SECONDS, null)) {
+                        if (!chunked.isClosed()
+                                && !writer.suspend(AsyncResponder.NO_TIMEOUT, TimeUnit.SECONDS, null)) {
                             LOGGER.fine(LocalizationMessages.ERROR_SUSPENDING_CHUNKED_OUTPUT_RESPONSE());
                         }
                     }