Add spring-core regression test for gh-37159 SubscriberInputStream interrupt fix

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;