Uncomment buffer leak tests in DataBufferUtils

Following a response on
reactor/reactor-core#1634 apparently
FluxSink is respecting doOnDiscard but there is a race condition
in DataBufferUtils with file reading.

This commit makes two changes:
1) Add more checks for onDispose to detect cancellation signals
as well as to deal with potentially concurrent such signal.
2) Do not close the channel through the Flux.using callback but
rather allow the current I/O callback to take place and only then
close the channel or else the buffer is left hanging.

Despite this tests still can fail due to a suspected issue in Reactor
itself with the doOnDiscard callback for FluxSink. That's tracked under
the same issue reactor/reactor-core#1634
and for now the use of DefaultDataBufferFactory is enforced as a
workaround until the issue is resolved.

See gh-22107

Author: rstoyanchev

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

@@ -57,6 +57,12 @@ public abstract class DataBufferUtils {
 
 	private static final Consumer<DataBuffer> RELEASE_CONSUMER = DataBufferUtils::release;
 
+	/**
+	 * Workaround to disable use of pooled buffers:
+	 * https://github.com/reactor/reactor-core/issues/1634
+	 */
+	private static final DataBufferFactory defaultDataBufferFactory = new DefaultDataBufferFactory();
+
 
 	//---------------------------------------------------------------------
 	// Reading
@@ -132,16 +138,21 @@ public static Flux<DataBuffer> readAsynchronousFileChannel(Callable<Asynchronous
 		Assert.isTrue(position >= 0, "'position' must be >= 0");
 		Assert.isTrue(bufferSize > 0, "'bufferSize' must be > 0");
 
+		DataBufferFactory bufferFactoryToUse = defaultDataBufferFactory;
+
 		Flux<DataBuffer> flux = Flux.using(channelSupplier,
 				channel -> Flux.create(sink -> {
 					ReadCompletionHandler handler =
-							new ReadCompletionHandler(channel, sink, position, bufferFactory, bufferSize);
-					DataBuffer dataBuffer = bufferFactory.allocateBuffer(bufferSize);
+							new ReadCompletionHandler(channel, sink, position, bufferFactoryToUse, bufferSize);
+					DataBuffer dataBuffer = bufferFactoryToUse.allocateBuffer(bufferSize);
 					ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize);
 					channel.read(byteBuffer, position, dataBuffer, handler);
 					sink.onDispose(handler::dispose);
 				}),
-				DataBufferUtils::closeChannel);
+				channel -> {
+					// Do not close channel from here, rather wait for the current read callback
+					// and then complete after releasing the DataBuffer.
+				});
 
 		return flux.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
 	}
@@ -505,26 +516,40 @@ public ReadCompletionHandler(AsynchronousFileChannel channel,
 
 		@Override
 		public void completed(Integer read, DataBuffer dataBuffer) {
-			if (read != -1) {
+			if (read != -1 && !this.disposed.get()) {
 				long pos = this.position.addAndGet(read);
 				dataBuffer.writePosition(read);
 				this.sink.next(dataBuffer);
-				if (!this.disposed.get()) {
+				// It's possible for cancellation to happen right before the push into the sink
+				if (this.disposed.get()) {
+					// TODO:
+					// This is not ideal since we already passed the buffer into the sink and
+					// releasing may cause something reading to fail. Maybe we won't have to
+					// do this after https://github.com/reactor/reactor-core/issues/1634
+					complete(dataBuffer);
+				}
+				else {
 					DataBuffer newDataBuffer = this.dataBufferFactory.allocateBuffer(this.bufferSize);
 					ByteBuffer newByteBuffer = newDataBuffer.asByteBuffer(0, this.bufferSize);
 					this.channel.read(newByteBuffer, pos, newDataBuffer, this);
 				}
 			}
 			else {
-				release(dataBuffer);
-				this.sink.complete();
+				complete(dataBuffer);
 			}
 		}
 
+		private void complete(DataBuffer dataBuffer) {
+			release(dataBuffer);
+			this.sink.complete();
+			closeChannel(this.channel);
+		}
+
 		@Override
 		public void failed(Throwable exc, DataBuffer dataBuffer) {
 			release(dataBuffer);
 			this.sink.error(exc);
+			closeChannel(this.channel);
 		}
 
 		public void dispose() {

spring-core/src/test/java/org/springframework/core/codec/ResourceRegionEncoderTests.java

@@ -80,8 +80,7 @@ public void shouldEncodeResourceRegionFileResource() throws Exception {
 				.expectComplete()
 				.verify();
 
-		// TODO: https://github.com/reactor/reactor-core/issues/1634
-		// this.bufferFactory.checkForLeaks();
+		 this.bufferFactory.checkForLeaks();
 	}
 
 	@Test
@@ -122,11 +121,10 @@ public void shouldEncodeMultipleResourceRegionsFileResource() {
 				.expectComplete()
 				.verify();
 
-		// TODO: https://github.com/reactor/reactor-core/issues/1634
-		// this.bufferFactory.checkForLeaks();
+		 this.bufferFactory.checkForLeaks();
 	}
 
-	@Test // gh-
+	@Test // gh-22107
 	public void cancelWithoutDemandForMultipleResourceRegions() {
 		Resource resource = new ClassPathResource("ResourceRegionEncoderTests.txt", getClass());
 		Flux<ResourceRegion> regions = Flux.just(
@@ -173,8 +171,7 @@ public void nonExisting() {
 				.expectError(EncodingException.class)
 				.verify();
 
-		// TODO: https://github.com/reactor/reactor-core/issues/1634
-		// this.bufferFactory.checkForLeaks();
+		 this.bufferFactory.checkForLeaks();
 	}
 
 	protected Consumer<DataBuffer> stringConsumer(String expected) {

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

@@ -135,6 +135,7 @@ private void verifyAllocations() {
 						catch (InterruptedException ex) {
 							// ignore
 						}
+						continue;
 					}
 					assertEquals("ByteBuf Leak: " + total + " unreleased allocations", 0, total);
 				}

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

@@ -187,8 +187,6 @@ public void readAsynchronousFileChannelCancel() throws Exception {
 				.verify();
 	}
 
-	// TODO: Remove ignore after https://github.com/reactor/reactor-core/issues/1634
-	@Ignore
 	@Test // gh-22107
 	public void readAsynchronousFileChannelCancelWithoutDemand() throws Exception {
 		URI uri = this.resource.getURI();

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

@@ -85,6 +85,7 @@ public void checkForLeaks() {
 				catch (InterruptedException ex) {
 					// ignore
 				}
+				continue;
 			}
 			List<AssertionError> errors = this.created.stream()
 					.filter(LeakAwareDataBuffer::isAllocated)