Flux/Mono.usingWhen late resource memory leak (reactor#3695)

UsingWhen method may lead to a memory or resource leak of allocation was finished after the main pipeline was cancelled. This change assures that asyncCancel is always called even if deferred.

Fixes reactor#3695.

Author: Maksym Lozbin

reactor-core/src/main/java/reactor/core/publisher/FluxUsingWhen.java

@@ -25,7 +25,6 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
-
 import reactor.core.CoreSubscriber;
 import reactor.core.Exceptions;
 import reactor.core.Fuseable.ConditionalSubscriber;
@@ -242,15 +241,6 @@ public void onSubscribe(Subscription s) {
 			}
 		}
 
-		@Override
-		public void cancel() {
-			if (!resourceProvided) {
-				resourceSubscription.cancel();
-			}
-
-			super.cancel();
-		}
-
 		@Override
 		public Object scanUnsafe(Attr key) {
 			if (key == Attr.PARENT) return resourceSubscription;

reactor-core/src/main/java/reactor/core/publisher/MonoUsingWhen.java

@@ -24,7 +24,6 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
-
 import reactor.core.CoreSubscriber;
 import reactor.core.Exceptions;
 import reactor.core.publisher.FluxUsingWhen.UsingWhenSubscriber;
@@ -222,15 +221,6 @@ public void onSubscribe(Subscription s) {
 			}
 		}
 
-		@Override
-		public void cancel() {
-			if (!resourceProvided) {
-				resourceSubscription.cancel();
-			}
-
-			super.cancel();
-		}
-
 		@Override
 		public Object scanUnsafe(Attr key) {
 			if (key == Attr.PARENT) return resourceSubscription;

reactor-core/src/test/java/reactor/core/publisher/FluxUsingWhenTest.java

@@ -22,6 +22,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.LongAdder;
+import java.util.concurrent.locks.LockSupport;
 import java.util.function.Function;
 import java.util.logging.Level;
 import java.util.stream.Stream;
@@ -33,20 +34,19 @@
 import org.mockito.Mockito;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
-
 import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
 import reactor.core.Fuseable;
 import reactor.core.Scannable.Attr;
 import reactor.core.TestLoggerExtension;
 import reactor.core.publisher.FluxUsingWhen.ResourceSubscriber;
 import reactor.core.publisher.FluxUsingWhen.UsingWhenSubscriber;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.ParameterizedTestWithName;
 import reactor.test.StepVerifier;
 import reactor.test.publisher.PublisherProbe;
 import reactor.test.publisher.TestPublisher;
 import reactor.test.util.TestLogger;
-import reactor.util.Loggers;
 import reactor.util.annotation.Nullable;
 import reactor.util.context.Context;
 import reactor.util.function.Tuple2;
@@ -705,6 +705,43 @@ public void apiCancel(Flux<String> transactionToCancel) {
 				.matches(tr -> tr.cancelProbe.wasSubscribed(), "cancel method used");
 	}
 
+	@ParameterizedTestWithName
+	@MethodSource("sourcesFullTransaction")
+	public void lateResourcePublisherCleanupIsDeferredOnCancel(Flux<String> transactionToCancel)
+			throws InterruptedException {
+		AtomicReference<TestResource> ref = new AtomicReference<>();
+		CountDownLatch resourceSubscribeLatch = new CountDownLatch(1);
+		CountDownLatch resourceCancelLatch = new CountDownLatch(1);
+		Flux<String> flux = Flux.usingWhen(Mono.fromCallable(() -> {
+					LockSupport.parkNanos(Duration.ofMillis(100).toNanos());
+					TestResource testResource = new TestResource();
+					ref.set(testResource);
+					resourceSubscribeLatch.countDown();
+					return testResource;
+				}).subscribeOn(Schedulers.single()),
+				d -> transactionToCancel,
+				TestResource::commit,
+				TestResource::rollback,
+				testResource -> testResource.cancel()
+				                            .doOnSubscribe(unused -> resourceCancelLatch.countDown()));
+
+		StepVerifier.create(flux.take(Duration.ofMillis(10)), 1)
+		            .verifyComplete();
+
+		assertThat(resourceSubscribeLatch.await(1, TimeUnit.SECONDS))
+				.as("Resource create subscribed")
+				.isTrue();
+		assertThat(resourceCancelLatch.await(1, TimeUnit.SECONDS))
+				.as("Resource cancel subscribed")
+				.isTrue();
+
+		assertThat(ref.get())
+				.isNotNull()
+				.matches(tr -> !tr.commitProbe.wasSubscribed(), "no commit")
+				.matches(tr -> !tr.rollbackProbe.wasSubscribed(), "no rollback")
+				.matches(tr -> tr.cancelProbe.wasSubscribed(), "cancel method used");
+	}
+
 	@ParameterizedTestWithName
 	@MethodSource("sourcesFullTransaction")
 	@TestLoggerExtension.Redirect