ThrottledInputStream.throttle() doesn't preserve the original

InterruptedException as the cause of its InterruptedIOException

ThrottledInputStream.throttle() now restores the current thread's
interrupt flag when catching InterruptedException.

Author: garydgregory

src/changes/changes.xml

@@ -56,6 +56,7 @@ The <action> type attribute can be add,update,fix,remove.
       <action type="fix" dev="ggregory" due-to="Gary Gregory">FileCleaningTracker now restores the current thread's interrupt flag when catching InterruptedException.</action>
       <action type="fix" dev="ggregory" due-to="Gary Gregory">ThreadMonitor.run() now restores the current thread's interrupt flag when catching InterruptedException.</action>
       <action type="fix" dev="ggregory" due-to="Gary Gregory">ThrottledInputStream.throttle() now restores the current thread's interrupt flag when catching InterruptedException.</action>
+      <action type="fix" dev="ggregory" due-to="Gary Gregory">ThrottledInputStream.throttle() doesn't preserve the original InterruptedException as the cause of its InterruptedIOException.</action>
       <!-- ADD -->
       <action type="add" dev="ggregory" due-to="Gary Gregory, Piotr P. Karwasz">Add and use IOUtils.closeQuietlySuppress(Closeable, Throwable) #818.</action>
       <!-- UPDATE -->

src/main/java/org/apache/commons/io/input/Input.java

@@ -18,11 +18,12 @@
 package org.apache.commons.io.input;
 
 import java.io.IOException;
+import java.io.InterruptedIOException;
 
 /**
  * Package-wide internals for input.
  */
-class Input {
+final class Input {
 
     /**
      * Throws an IOException on false input.
@@ -36,4 +37,21 @@ static void checkOpen(final boolean isOpen) throws IOException {
         }
     }
 
+    /**
+     * Converts an InterruptedException to an InterruptedIOException.
+     * <p>
+     * The cause of the returned InterruptedIOException is set to the original.
+     * </p>
+     *
+     * @param e The InterruptedException to convert.
+     * @return The converted InterruptedIOException.
+     * @see InterruptedIOException
+     * @see Throwable#initCause(Throwable)
+     * @see Throwable#getCause()
+     */
+    static InterruptedIOException toInterruptedIOException(final InterruptedException e) {
+        final InterruptedIOException iio = new InterruptedIOException(e.getMessage());
+        iio.initCause(e);
+        return iio;
+    }
 }

src/main/java/org/apache/commons/io/input/ReadAheadInputStream.java

@@ -20,7 +20,6 @@
 import java.io.FilterInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.io.InterruptedIOException;
 import java.nio.ByteBuffer;
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
@@ -299,9 +298,7 @@ public void close() throws IOException {
                 executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
             } catch (final InterruptedException e) {
                 Thread.currentThread().interrupt();
-                final InterruptedIOException iio = new InterruptedIOException(e.getMessage());
-                iio.initCause(e);
-                throw iio;
+                throw Input.toInterruptedIOException(e);
             } finally {
                 if (isSafeToCloseUnderlyingInputStream) {
                     super.close();
@@ -559,9 +556,7 @@ private void waitForAsyncReadComplete() throws IOException {
             }
         } catch (final InterruptedException e) {
             Thread.currentThread().interrupt();
-            final InterruptedIOException iio = new InterruptedIOException(e.getMessage());
-            iio.initCause(e);
-            throw iio;
+            throw Input.toInterruptedIOException(e);
         } finally {
             try {
                 isWaiting.set(false);

src/main/java/org/apache/commons/io/input/ThrottledInputStream.java

@@ -245,21 +245,38 @@ private long getBytesPerSecond() {
         return getByteCount() / elapsedSeconds;
     }
 
-    // package private for testing.
+    /**
+     * Gets the maximum bytes per second.
+     * <p>
+     * Package private for testing.
+     * </p>
+     *
+     * @return The maximum bytes per second.
+     */
     double getMaxBytesPerSecond() {
         return maxBytesPerSecond;
     }
 
-    private long getSleepMillis() {
+    /**
+     * Gets the number of milliseconds to sleep to match to the maximum bytes per second.
+     * <p>
+     * Package private for testing.
+     * </p>
+     *
+     * @return the number of milliseconds to sleep to match to the maximum bytes per second.
+     */
+    long getSleepMillis() {
         return toSleepMillis(getByteCount(), System.currentTimeMillis() - startTime, maxBytesPerSecond);
     }
 
     /**
      * Gets the total duration spent in sleep.
+     * <p>
+     * Package private for testing
+     * </p>
      *
      * @return Duration spent in sleep.
      */
-    // package private for testing
     Duration getTotalSleepDuration() {
         return totalSleepDuration;
     }
@@ -272,7 +289,7 @@ private void throttle() throws InterruptedIOException {
                 TimeUnit.MILLISECONDS.sleep(sleepMillis);
             } catch (final InterruptedException e) {
                 Thread.currentThread().interrupt();
-                throw new InterruptedIOException("Thread aborted");
+                throw Input.toInterruptedIOException(e);
             }
         }
     }

src/test/java/org/apache/commons/io/input/ThrottledInputStreamTest.java

@@ -18,12 +18,16 @@
 package org.apache.commons.io.input;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InterruptedIOException;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -188,4 +192,20 @@ void testGet() throws IOException {
         }
     }
 
+    @Test
+    synchronized void testReadInterrupt() throws IOException {
+        try (ThrottledInputStream inputStream = ThrottledInputStream.builder()
+                // @formatter:off
+                .setInputStream(createOriginInputStream())
+                .setMaxBytes(1, ChronoUnit.HOURS)
+                .get()) {
+                // @formatter:on
+            final ThrottledInputStream spy = spy(inputStream);
+            when(spy.getSleepMillis()).thenReturn(1L);
+            Thread.currentThread().interrupt();
+            assertInstanceOf(InterruptedException.class, assertThrows(InterruptedIOException.class, spy::read).getCause());
+            assertTrue(Thread.interrupted());
+        }
+    }
+
 }