Merge branch '7.0.x'

This commit is contained in:
Sam Brannen
2026-08-18 17:14:31 +02:00
2 changed files with 73 additions and 9 deletions
@@ -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();
@@ -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;