Polishing contribution

This commit adds further fixes in the same area, since there were
similar bugs in the WriteCompletionHandler:
* databuffers were not always emitted when fully read in the onNext hook
* on completion, the iterator was closed too early, before it was fully
  read
* on completion, writing the next bytebuffers from the iterator would
  always reuse the first one and not update the attachment

Closes gh-36714

Author: bclozel

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

@@ -1215,6 +1215,11 @@ protected void hookOnNext(DataBuffer dataBuffer) {
 					failed(ex, attachment);
 				}
 			}
+			else {
+				iterator.close();
+				this.sink.next(dataBuffer);
+				request(1);
+			}
 		}
 
 		@Override
@@ -1238,7 +1243,6 @@ protected void hookOnComplete() {
 		@Override
 		public void completed(Integer written, Attachment attachment) {
 			DataBuffer.ByteBufferIterator iterator = attachment.iterator();
-			iterator.close();
 
 			long pos = this.position.addAndGet(written);
 			ByteBuffer byteBuffer = attachment.byteBuffer();
@@ -1248,9 +1252,11 @@ public void completed(Integer written, Attachment attachment) {
 			}
 			else if (iterator.hasNext()) {
 				ByteBuffer next = iterator.next();
-				this.channel.write(next, pos, attachment, this);
+				Attachment nextAttachment = new Attachment(next, attachment.dataBuffer(), iterator);
+				this.channel.write(next, pos, nextAttachment, this);
 			}
 			else {
+				iterator.close();
 				this.sink.next(attachment.dataBuffer());
 				this.writing.set(false);
 

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

@@ -466,6 +466,27 @@ private void verifyWrittenData(Flux<DataBuffer> writeResult) throws IOException
 		assertThat(result).isEqualTo("foobarbazqux");
 	}
 
+	@ParameterizedDataBufferAllocatingTest
+	void writeAsynchronousFileChannelWithJoinedBuffer(DataBufferFactory bufferFactory) throws Exception {
+		super.bufferFactory = bufferFactory;
+
+		DataBuffer foo = stringBuffer("foo");
+		DataBuffer bar = stringBuffer("bar");
+		DataBuffer joined = bufferFactory.join(List.of(foo, bar));
+
+		AsynchronousFileChannel channel = AsynchronousFileChannel.open(tempFile, StandardOpenOption.WRITE);
+
+		Flux<DataBuffer> writeResult = DataBufferUtils.write(Flux.just(joined), channel);
+		StepVerifier.create(writeResult)
+				.consumeNextWith(stringConsumer("foobar"))
+				.verifyComplete();
+
+		String result = String.join("", Files.readAllLines(tempFile));
+
+		assertThat(result).isEqualTo("foobar");
+		channel.close();
+	}
+
 	@ParameterizedDataBufferAllocatingTest
 	void writeAsynchronousFileChannelErrorInFlux(DataBufferFactory bufferFactory) throws Exception {
 		super.bufferFactory = bufferFactory;