diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java index 860c83c55f2..6b9b1014ce9 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java @@ -1223,13 +1223,7 @@ public abstract class DataBufferUtils { long pos = this.position.get(); Attachment attachment = new Attachment(byteBuffer, dataBuffer, iterator); this.writing.set(true); - try { - this.channel.write(byteBuffer, pos, attachment, this); - } - catch (RuntimeException ex) { - // If the exception escapes, route it to the failure handler - failed(ex, attachment); - } + write(byteBuffer, pos, attachment); } else { iterator.close(); @@ -1264,12 +1258,12 @@ public abstract class DataBufferUtils { ByteBuffer byteBuffer = attachment.byteBuffer(); if (byteBuffer.hasRemaining()) { - this.channel.write(byteBuffer, pos, attachment, this); + write(byteBuffer, pos, attachment); } else if (iterator.hasNext()) { ByteBuffer next = iterator.next(); Attachment nextAttachment = new Attachment(next, attachment.dataBuffer(), iterator); - this.channel.write(next, pos, nextAttachment, this); + write(next, pos, nextAttachment); } else { iterator.close(); @@ -1289,6 +1283,17 @@ public abstract class DataBufferUtils { } } + private void write(ByteBuffer byteBuffer, long pos, Attachment attachment) { + try { + this.channel.write(byteBuffer, pos, attachment, this); + } + catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + // If the exception escapes, route it to the failure handler + failed(ex, attachment); + } + } + @Override public void failed(Throwable ex, Attachment attachment) { attachment.iterator().close(); 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 238e64040bf..8885adef9a0 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 @@ -611,6 +611,65 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests { channel.close(); } + @Test // gh-37145 + void writeAsynchronousFileChannelWriteThrowsErrorSynchronously() { + super.bufferFactory = new DefaultDataBufferFactory(); + + DataBuffer foo = stringBuffer("foo"); + Flux flux = Flux.just(foo); + + // Since AssertionError is not a JVM-fatal exception, Exceptions.throwIfFatal() + // lets it through to the failure handler. + AsynchronousFileChannel channel = mock(); + willThrow(new AssertionError("simulated synchronous failure")) + .given(channel).write(any(), anyLong(), any(), any()); + + Flux writeResult = DataBufferUtils.write(flux, channel); + + StepVerifier.create(writeResult) + .consumeNextWith(stringConsumer("foo")) + .expectError(AssertionError.class) + .verify(Duration.ofSeconds(3)); + } + + @Test // gh-37145 + void writeAsynchronousFileChannelWriteThrowsErrorSynchronouslyFromCompletionThread() { + super.bufferFactory = new DefaultDataBufferFactory(); + + DataBuffer foo = stringBuffer("foo"); + Flux flux = Flux.just(foo); + + // Real AsynchronousFileChannel implementations invoke the CompletionHandler on a + // separate thread, not the calling thread. If the OS only writes part of the + // buffer, WriteCompletionHandler#completed recursively calls channel.write(...) + // again for the remainder - on that other thread, not the original caller. + var executor = Executors.newSingleThreadExecutor(); + try { + AsynchronousFileChannel channel = mock(); + willAnswer(invocation -> { + ByteBuffer buffer = invocation.getArgument(0); + Object attachment = invocation.getArgument(2); + CompletionHandler completionHandler = invocation.getArgument(3); + // Simulate a partial write (1 of 3 bytes) so that completed() has to + // recursively write the remainder. + buffer.position(buffer.position() + 1); + executor.submit(() -> completionHandler.completed(1, attachment)); + return null; + }).willThrow(new AssertionError("simulated synchronous failure")) + .given(channel).write(any(), anyLong(), any(), any()); + + Flux writeResult = DataBufferUtils.write(flux, channel); + + StepVerifier.create(writeResult) + .consumeNextWith(stringConsumer("foo")) + .expectError(AssertionError.class) + .verify(Duration.ofSeconds(3)); + } + finally { + executor.shutdown(); + } + } + @ParameterizedDataBufferAllocatingTest void writeAsynchronousFileChannelCanceled(DataBufferFactory bufferFactory) throws Exception { super.bufferFactory = bufferFactory;