From a4720ccf771211cabbd291385db1eef8ddd16a43 Mon Sep 17 00:00:00 2001 From: Sam Brannen <104798+sbrannen@users.noreply.github.com> Date: Tue, 18 Aug 2026 17:08:06 +0200 Subject: [PATCH] Guard all AsynchronousFileChannel#write call sites in DataBufferUtils Prior to this commit, only the very first AsynchronousFileChannel#write call in DataBufferUtils$WriteCompletionHandler#hookOnNext(DataBuffer) was guarded against exceptions escaping synchronously, and even then only via `catch (RuntimeException ex)`, per the original fix for gh-36184. While widening that guard to match the read side's `catch (Throwable ex)` combined with `Exceptions.throwIfFatal(ex)` (see gh-37143), we discovered that completed(Integer, Attachment) contains two more direct `this.channel.write(...)` calls -- for continuing a partial write and for advancing to the next ByteBuffer within the same DataBuffer's iterator -- neither of which was guarded at all. Since completed() is invoked by the channel's own completion callback, typically on a different thread than the one that issued the original write, a synchronous exception escaping either of those calls has no path back to the FluxSink, and the resulting Flux hangs indefinitely, exactly as described in gh-37143, for any write that receives a partial OS write or spans multiple ByteBuffers. To address that, this commit extracts a private write(ByteBuffer, long, Attachment) helper that wraps the channel.write(...) call with a try/catch block, routing any non-fatal Throwable -- via Exceptions.throwIfFatal() -- to the existing failed(Throwable, Attachment) handler. All three call sites (hookOnNext() and both branches in completed()) now go through this helper, ensuring the Flux always terminates with a proper error signal instead of hanging silently, regardless of which write attempt fails or which thread it fails on. See gh-36184 See gh-37143 Closes gh-37145 --- .../core/io/buffer/DataBufferUtils.java | 23 +++++--- .../core/io/buffer/DataBufferUtilsTests.java | 59 +++++++++++++++++++ 2 files changed, 73 insertions(+), 9 deletions(-) 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;