Implemented a workaround for JRFCAF-1344.

-  Underlying entity stream close() implementation may be thread-unsafe and as such the close() may result in an IOException at the socket input stream level, if the close() gets called at once from multiple threads somehow.

Change-Id: I6b3d05504d182515d001cc6244e8f0d8eee9ce36
Signed-off-by: Marek Potociar <[email protected]>

Author: Marek Potociar

core-common/src/main/java/org/glassfish/jersey/message/internal/EntityInputStream.java

@@ -42,6 +42,9 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.PushbackInputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 import javax.ws.rs.ProcessingException;
 
@@ -57,8 +60,10 @@
  * @author Marek Potociar (marek.potociar at oracle.com)
  */
 public class EntityInputStream extends InputStream {
-    private InputStream input;
-    private boolean closed;
+    private static final Logger LOGGER = Logger.getLogger(EntityInputStream.class.getName());
+
+    private volatile InputStream input;
+    private final AtomicBoolean closed = new AtomicBoolean(false);
 
     /**
      * Create an entity input stream instance wrapping the original input stream.
@@ -144,21 +149,24 @@ public void reset() {
      * {@inheritDoc}
      * <p>
      * The method is customized to not throw an {@link IOException} if the close operation fails. Instead,
-     * a runtime {@link javax.ws.rs.ProcessingException} is thrown.
+     * a warning message is logged.
      * </p>
-     *
-     * @throws javax.ws.rs.ProcessingException
-     *          in case the close operation on the underlying entity input stream failed.
      */
     @Override
     public void close() throws ProcessingException {
-        if (!closed && input != null) {
+        final InputStream in = input;
+        if (in == null) {
+            return;
+        }
+        if (closed.compareAndSet(false, true)) {
+            // Workaround for JRFCAF-1344: Underlying stream close() may be thread-unsafe
+            // and as such the close() may result in an IOException at the socket input stream level,
+            // if the close() gets called at once from multiple threads somehow.
             try {
-                input.close();
+                in.close();
             } catch (IOException ex) {
-                throw new ProcessingException(LocalizationMessages.MESSAGE_CONTENT_INPUT_STREAM_CLOSE_FAILED(), ex);
-            } finally {
-                closed = true;
+                // This means that the underlying socket stream got closed by other thread somehow
+                LOGGER.log(Level.FINE, LocalizationMessages.MESSAGE_CONTENT_INPUT_STREAM_CLOSE_FAILED(), ex);
             }
         }
     }
@@ -174,36 +182,37 @@ public void close() throws ProcessingException {
     public boolean isEmpty() {
         ensureNotClosed();
 
-        if (input == null) {
+        final InputStream in = input;
+        if (in == null) {
             return true;
         }
 
         try {
             // Try #markSupported first - #available on WLS waits until socked timeout is reached when chunked encoding is used.
-            if (input.markSupported()) {
-                input.mark(1);
-                int i = input.read();
-                input.reset();
+            if (in.markSupported()) {
+                in.mark(1);
+                int i = in.read();
+                in.reset();
                 return i == -1;
             } else {
                 try {
-                    if (input.available() > 0) {
+                    if (in.available() > 0) {
                         return false;
                     }
                 } catch (IOException ioe) {
                     // NOOP. Try other approaches as this can fail on WLS.
                 }
 
-                int b = input.read();
+                int b = in.read();
                 if (b == -1) {
                     return true;
                 }
 
                 PushbackInputStream pbis;
-                if (input instanceof PushbackInputStream) {
-                    pbis = (PushbackInputStream) input;
+                if (in instanceof PushbackInputStream) {
+                    pbis = (PushbackInputStream) in;
                 } else {
-                    pbis = new PushbackInputStream(input, 1);
+                    pbis = new PushbackInputStream(in, 1);
                     input = pbis;
                 }
                 pbis.unread(b);
@@ -221,7 +230,7 @@ public boolean isEmpty() {
      * @throws IllegalStateException in case the entity input stream has been closed.
      */
     public void ensureNotClosed() throws IllegalStateException {
-        if (closed) {
+        if (closed.get()) {
             throw new IllegalStateException(LocalizationMessages.ERROR_ENTITY_STREAM_CLOSED());
         }
     }
@@ -232,7 +241,7 @@ public void ensureNotClosed() throws IllegalStateException {
      * @return {@code true} if the stream has been closed, {@code false} otherwise.
      */
     public boolean isClosed() {
-        return closed;
+        return closed.get();
     }
 
     /**