From 58f5a9e81c51f603fa980f0783bf9473b595af8e Mon Sep 17 00:00:00 2001 From: "jaipilot[bot]" <273169020+jaipilot[bot]@users.noreply.github.com> Date: Sat, 22 Aug 2026 12:33:47 +0000 Subject: [PATCH] Consolidate duplicated read() error handling and add spring-core interrupt test for SubscriberInputStream --- .../core/io/buffer/SubscriberInputStream.java | 32 +++++++------ .../core/io/buffer/DataBufferUtilsTests.java | 46 +++++++++++++++++++ .../http/client/SubscriberInputStream.java | 32 +++++++------ 3 files changed, 82 insertions(+), 28 deletions(-) diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java b/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java index db10836dd87c..f825419de311 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java @@ -226,13 +226,7 @@ else if (next == CLOSED) { return next.read() & 0xFF; } catch (Throwable ex) { - this.closed = true; - requiredSubscriber().cancel(); - cleanAndFinalize(); - if (ex instanceof IOException ioEx) { - throw ioEx; - } - throw Exceptions.propagate(ex); + throw handleReadError(ex); } finally { this.lock.unlock(); @@ -285,19 +279,29 @@ else if (next == CLOSED) { return len; } catch (Throwable ex) { - this.closed = true; - requiredSubscriber().cancel(); - cleanAndFinalize(); - if (ex instanceof IOException ioEx) { - throw ioEx; - } - throw Exceptions.propagate(ex); + throw handleReadError(ex); } finally { this.lock.unlock(); } } + /** + * Mark this stream closed, cancel the subscription, release stored buffers, + * and translate the given exception into one that a {@code read} method can + * throw: an {@link IOException} is rethrown as-is, anything else is wrapped + * via {@link Exceptions#propagate}. + */ + private RuntimeException handleReadError(Throwable ex) throws IOException { + this.closed = true; + requiredSubscriber().cancel(); + cleanAndFinalize(); + if (ex instanceof IOException ioEx) { + throw ioEx; + } + throw Exceptions.propagate(ex); + } + private DataBuffer getNextOrAwait() throws InterruptedIOException { if (this.available == null || this.available.readableByteCount() == 0) { discard(this.available); diff --git a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java index 8885adef9a09..ef447217eb5a 100644 --- a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java +++ b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java @@ -18,6 +18,7 @@ import java.io.IOException; import java.io.InputStream; +import java.io.InterruptedIOException; import java.io.OutputStream; import java.io.OutputStreamWriter; import java.net.URI; @@ -39,6 +40,8 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; @@ -1049,6 +1052,49 @@ void inputStreamSubscriberClose(DataBufferFactory bufferFactory) throws Interrup } } + @Test // gh-37159 + void inputStreamSubscriberInterruptWhileAwaitingData() throws InterruptedException { + CountDownLatch reading = new CountDownLatch(1); + AtomicReference savedEx = new AtomicReference<>(); + AtomicBoolean interruptStatus = new AtomicBoolean(); + + // A publisher that never emits, so that read() parks in await() + Publisher publisher = subscriber -> subscriber.onSubscribe(new Subscription() { + @Override + public void request(long n) { + } + @Override + public void cancel() { + } + }); + + Thread reader = new Thread(() -> { + try (InputStream in = DataBufferUtils.subscriberInputStream(publisher, 1)) { + reading.countDown(); + in.read(); + } + catch (Throwable ex) { + savedEx.set(ex); + interruptStatus.set(Thread.currentThread().isInterrupted()); + } + }); + reader.start(); + + reading.await(); + for (int i = 0; i < 100 && reader.getState() != Thread.State.WAITING; i++) { + Thread.sleep(20); + } + assertThat(reader.getState()).isEqualTo(Thread.State.WAITING); + + reader.interrupt(); + reader.join(5000); + + assertThat(reader.isAlive()).as("read() did not return after interrupt").isFalse(); + assertThat(savedEx.get()).isInstanceOf(InterruptedIOException.class) + .hasMessage("Interrupted while awaiting data"); + assertThat(interruptStatus).as("interrupt status restored").isTrue(); + } + @ParameterizedDataBufferAllocatingTest void readAndWriteByteChannel(DataBufferFactory bufferFactory) throws Exception { super.bufferFactory = bufferFactory; diff --git a/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java b/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java index bd60b5eb2712..62916d960b70 100644 --- a/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java +++ b/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java @@ -247,13 +247,7 @@ else if (next == CLOSED) { return next[this.position++] & 0xFF; } catch (Throwable ex) { - this.closed = true; - requiredSubscriber().cancel(); - cleanAndFinalize(); - if (ex instanceof IOException ioEx) { - throw ioEx; - } - throw Exceptions.propagate(ex); + throw handleReadError(ex); } finally { this.lock.unlock(); @@ -308,19 +302,29 @@ else if (next == CLOSED) { return len; } catch (Throwable ex) { - this.closed = true; - requiredSubscriber().cancel(); - cleanAndFinalize(); - if (ex instanceof IOException ioEx) { - throw ioEx; - } - throw Exceptions.propagate(ex); + throw handleReadError(ex); } finally { this.lock.unlock(); } } + /** + * Mark this stream closed, cancel the subscription, release stored buffers, + * and translate the given exception into one that a {@code read} method can + * throw: an {@link IOException} is rethrown as-is, anything else is wrapped + * via {@link Exceptions#propagate}. + */ + private RuntimeException handleReadError(Throwable ex) throws IOException { + this.closed = true; + requiredSubscriber().cancel(); + cleanAndFinalize(); + if (ex instanceof IOException ioEx) { + throw ioEx; + } + throw Exceptions.propagate(ex); + } + byte[] getNextOrAwait() throws InterruptedIOException { if (this.available == null || this.available.length - this.position == 0) { this.available = null;