Consolidate duplicated read() error handling and add spring-core interrupt test for SubscriberInputStream

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);

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<Throwable> savedEx = new AtomicReference<>();
+		AtomicBoolean interruptStatus = new AtomicBoolean();
+
+		// A publisher that never emits, so that read() parks in await()
+		Publisher<DataBuffer> 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;

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;