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

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();

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<DataBuffer> 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<DataBuffer> 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<DataBuffer> 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<Integer, Object> 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<DataBuffer> 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;