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;