Guard against duplicate read completion callbacks

spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java

@@ -724,6 +724,10 @@ private void read() {
 
 		@Override
 		public void completed(Integer read, Attachment attachment) {
+			if (!attachment.markHandled()) {
+				return;
+			}
+
 			attachment.iterator().close();
 			DataBuffer dataBuffer = attachment.dataBuffer();
 
@@ -735,8 +739,8 @@ public void completed(Integer read, Attachment attachment) {
 
 			if (read == -1) {
 				release(dataBuffer);
-				closeChannel(this.channel);
 				this.state.set(State.DISPOSED);
+				closeChannel(this.channel);
 				this.sink.complete();
 				return;
 			}
@@ -759,19 +763,33 @@ public void completed(Integer read, Attachment attachment) {
 
 		@Override
 		public void failed(Throwable ex, Attachment attachment) {
+			if (!attachment.markHandled()) {
+				return;
+			}
+
 			attachment.iterator().close();
 			release(attachment.dataBuffer());
 
-			closeChannel(this.channel);
 			this.state.set(State.DISPOSED);
+			closeChannel(this.channel);
 			this.sink.error(ex);
 		}
 
 		private enum State {
 			IDLE, READING, DISPOSED
 		}
 
-		private record Attachment(DataBuffer dataBuffer, DataBuffer.ByteBufferIterator iterator) {}
+		private record Attachment(DataBuffer dataBuffer, DataBuffer.ByteBufferIterator iterator,
+				AtomicBoolean handled) {
+
+			Attachment(DataBuffer dataBuffer, DataBuffer.ByteBufferIterator iterator) {
+				this(dataBuffer, iterator, new AtomicBoolean());
+			}
+
+			boolean markHandled() {
+				return this.handled.compareAndSet(false, true);
+			}
+		}
 	}
 
 

spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java

@@ -22,6 +22,7 @@
 import java.io.OutputStreamWriter;
 import java.net.URI;
 import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
 import java.nio.channels.AsynchronousFileChannel;
 import java.nio.channels.CompletionHandler;
 import java.nio.channels.FileChannel;
@@ -39,6 +40,9 @@
 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 java.util.function.BiConsumer;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.PooledByteBufAllocator;
@@ -61,11 +65,14 @@
 import static org.assertj.core.api.Assertions.assertThatIOException;
 import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 /**
  * @author Arjen Poutsma
@@ -262,6 +269,58 @@ void readAsynchronousFileChannelReadThrowsSynchronouslyFromCompletionThread() th
 		}
 	}
 
+	@Test
+	void readAsynchronousFileChannelCompletedTwiceAfterEof() throws Exception {
+		readAsynchronousFileChannelWithDuplicateCallbackOnClose(
+				(handler, attachment) -> handler.completed(-1, attachment));
+	}
+
+	@Test
+	void readAsynchronousFileChannelFailedAfterEof() throws Exception {
+		readAsynchronousFileChannelWithDuplicateCallbackOnClose(
+				(handler, attachment) -> handler.failed(new AsynchronousCloseException(), attachment));
+	}
+
+	private void readAsynchronousFileChannelWithDuplicateCallbackOnClose(
+			BiConsumer<CompletionHandler<Integer, Object>, Object> duplicateCallback) throws Exception {
+
+		DataBuffer delegate = DefaultDataBufferFactory.sharedInstance.allocateBuffer(3);
+		PooledDataBuffer dataBuffer = mock();
+		given(dataBuffer.writableByteBuffers()).willAnswer(invocation -> delegate.writableByteBuffers());
+		given(dataBuffer.isAllocated()).willReturn(true);
+
+		DataBufferFactory factory = mock();
+		given(factory.allocateBuffer(anyInt())).willReturn(dataBuffer);
+
+		AtomicReference<CompletionHandler<Integer, Object>> handlerRef = new AtomicReference<>();
+		AtomicReference<Object> attachmentRef = new AtomicReference<>();
+
+		AtomicBoolean open = new AtomicBoolean(true);
+		AsynchronousFileChannel channel = mock();
+		given(channel.isOpen()).willAnswer(invocation -> open.get());
+		willAnswer(invocation -> {
+			open.set(false);
+			duplicateCallback.accept(handlerRef.get(), attachmentRef.get());
+			return null;
+		}).given(channel).close();
+		willAnswer(invocation -> {
+			Object attachment = invocation.getArgument(2);
+			CompletionHandler<Integer, Object> handler = invocation.getArgument(3);
+			handlerRef.set(handler);
+			attachmentRef.set(attachment);
+			handler.completed(-1, attachment);
+			return null;
+		}).given(channel).read(any(), anyLong(), any(), any());
+
+		Flux<DataBuffer> result = DataBufferUtils.readAsynchronousFileChannel(() -> channel, factory, 3);
+
+		StepVerifier.create(result)
+				.expectComplete()
+				.verify(Duration.ofSeconds(3));
+
+		verify(dataBuffer, times(1)).release();
+	}
+
 	@ParameterizedDataBufferAllocatingTest
 	void readAsynchronousFileChannelCancel(DataBufferFactory bufferFactory) throws Exception {
 		super.bufferFactory = bufferFactory;