Merge commit 'c628bdd3'

* commit 'c628bdd3':
  First pass at a fix for JENKINS-8592.

remoting/src/main/java/hudson/remoting/Pipe.java

@@ -23,6 +23,7 @@
  */
 package hudson.remoting;
 
+import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.ObjectInputStream;
@@ -92,7 +93,19 @@ public final class Pipe implements Serializable {
      * Gets the reading end of the pipe.
      */
     public InputStream getIn() {
-        return in;
+        return new FilterInputStream(in) {
+            @Override
+            public void close() throws IOException {
+                try {
+                    // Since closing the reading side does not stop the writing side. We read till the stream is done.
+                    final byte[] buffer = new byte[4096];
+                    while(read(buffer) != -1) {
+                    }
+                } finally {
+                    super.close();
+                }
+            }
+        };
     }
 
     /**

remoting/src/test/java/hudson/remoting/PipeTest.java

@@ -24,9 +24,13 @@
 package hudson.remoting;
 
 import hudson.remoting.ChannelRunner.InProcessCompatibilityMode;
+import hudson.remoting.FastPipedInputStream.ClosedBy;
 import junit.framework.Test;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.io.output.NullOutputStream;
+import org.jvnet.hudson.test.Bug;
+import org.jvnet.hudson.test.For;
+import org.jvnet.hudson.test.Url;
 
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
@@ -35,6 +39,9 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.io.Serializable;
 import java.util.Arrays;
+import java.util.logging.Handler;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
 
 /**
  * Test {@link Pipe}.
@@ -55,6 +62,60 @@ public class PipeTest extends RmiTestBase implements Serializable {
         System.out.println("result=" + r);
         assertEquals(5,r);
     }
+    
+    /**
+     * Helper class for testPartialReadQuietlyConsumesPipeOnClose.
+     */
+    private static class EventCounterHandler extends Handler {
+        int count = 0;
+        
+        @Override
+        public void publish(final LogRecord record) {
+            if (ProxyOutputStream.class.getName().equals(record.getLoggerName())) {
+                Throwable thrown = record.getThrown();
+                while (thrown != null) {
+                    if (thrown instanceof ClosedBy) {
+                        count += 1;
+                        break;
+                    }
+                    thrown = thrown.getCause();
+                }
+            }
+        }
+        
+        @Override
+        public void flush() {
+        }
+        
+        @Override
+        public void close() throws SecurityException {
+        }
+
+        public int getCount() {
+            return count;
+        }
+    }
+    
+    /**
+     * This test should be reproducing the initial bug as reported in JENKINS-8592. The assert in this test is not the
+     * best and is fragile, but it is the best I can come up with.
+     */
+    @Bug(8592)
+    @For(Pipe.class)
+    @Url("http://issues.jenkins-ci.org/browse/JENKINS-8592")
+    public void testPartialReadQuietlyConsumesPipeOnClose() throws Exception {
+        final EventCounterHandler handler = new EventCounterHandler();
+        final Logger logger = Logger.getLogger(ProxyOutputStream.class.getName());
+        logger.addHandler(handler);
+        final Pipe p = Pipe.createRemoteToLocal();
+        final Future<Integer> f = channel.callAsync(new WritingCallable(p));
+        final InputStream in = p.getIn();
+        assertEquals(in.read(), 0);
+        in.close();
+        f.get();
+        logger.removeHandler(handler);
+        assertEquals(0, handler.getCount());
+    }
 
     private static class WritingCallable implements Callable<Integer, IOException> {
         private final Pipe pipe;