Fix busy-spin in SubscriberInputStream await()

spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java

@@ -391,10 +391,23 @@ private void await() {
 				throw new IllegalStateException("Only one (Virtual)Thread can await!");
 			}
 
-			if (this.parkedThread.compareAndSet( null, toUnpark)) {
+			if (this.parkedThread.compareAndSet(null, toUnpark) || this.parkedThread.get() == toUnpark) {
 				LockSupport.park();
 				// we don't just break here because park() can wake up spuriously
 				// if we got a proper resume, get() == READY and the loop will quit above
+				// if we woke up spuriously, the reference is still ours and we park again
+			}
+
+			// park() also returns on interruption, without a resume() having set READY,
+			// in which case the loop could otherwise neither park again nor exit.
+			// Honor the cancellation, unless data arrived concurrently: then deliver it
+			// first and leave the thread interrupted for the next await() call.
+			if (Thread.interrupted()) {
+				if (this.parkedThread.get() != READY) {
+					this.parkedThread.lazySet(null);
+					throw new IllegalStateException("Interrupted while awaiting data");
+				}
+				Thread.currentThread().interrupt();
 			}
 		}
 		// clear the resume indicator so that the next await call will park without a resume()

spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java

@@ -420,10 +420,23 @@ private void await() {
 				throw new IllegalStateException("Only one (Virtual)Thread can await!");
 			}
 
-			if (this.parkedThread.compareAndSet( null, toUnpark)) {
+			if (this.parkedThread.compareAndSet(null, toUnpark) || this.parkedThread.get() == toUnpark) {
 				LockSupport.park();
 				// we don't just break here because park() can wake up spuriously
 				// if we got a proper resume, get() == READY and the loop will quit above
+				// if we woke up spuriously, the reference is still ours and we park again
+			}
+
+			// park() also returns on interruption, without a resume() having set READY,
+			// in which case the loop could otherwise neither park again nor exit.
+			// Honor the cancellation, unless data arrived concurrently: then deliver it
+			// first and leave the thread interrupted for the next await() call.
+			if (Thread.interrupted()) {
+				if (this.parkedThread.get() != READY) {
+					this.parkedThread.lazySet(null);
+					throw new IllegalStateException("Interrupted while awaiting data");
+				}
+				Thread.currentThread().interrupt();
 			}
 		}
 		// clear the resume indicator so that the next await call will park without a resume()

spring-web/src/test/java/org/springframework/http/client/SubscriberInputStreamTests.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Flow;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.jupiter.api.Test;
 
@@ -233,4 +234,44 @@ void mapperThrowsException() throws InterruptedException {
 		assertThat(savedEx).hasMessage("boom");
 	}
 
+	@Test // gh-37159
+	void interruptWhileAwaitingData() throws InterruptedException {
+		CountDownLatch reading = new CountDownLatch(1);
+		AtomicReference<Throwable> savedEx = new AtomicReference<>();
+
+		// A publisher that never emits, so that read() parks in await()
+		Flow.Publisher<byte[]> publisher = subscriber -> subscriber.onSubscribe(new Flow.Subscription() {
+			@Override
+			public void request(long n) {
+			}
+			@Override
+			public void cancel() {
+			}
+		});
+
+		Thread reader = new Thread(() -> {
+			try (SubscriberInputStream<byte[]> is = new SubscriberInputStream<>(s -> s, s -> {}, 1)) {
+				publisher.subscribe(is);
+				reading.countDown();
+				is.read();
+			}
+			catch (Throwable ex) {
+				savedEx.set(ex);
+			}
+		});
+		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()).hasMessage("Interrupted while awaiting data");
+	}
+
 }