verify inflight mono's cancelled for mono firstWithValue

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2020-2024 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,11 +20,13 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.NoSuchElementException;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.junit.jupiter.api.Test;
 
 import reactor.core.Exceptions;
 import reactor.core.Scannable;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 import reactor.test.publisher.TestPublisher;
 import reactor.test.subscriber.AssertSubscriber;
@@ -172,6 +174,39 @@ void cancelIsPropagated() {
 		pub2.assertMaxRequested(1);
 		pub2.assertWasCancelled();
 	}
+	@Test
+	void cancelInflightMono() {
+		AtomicLong counter = new AtomicLong();
+		Mono<Integer> inflightMono1 = Mono.fromCallable(() -> {
+			try {
+				Thread.sleep(200);
+			} catch (InterruptedException e) {
+				counter.incrementAndGet();
+			}
+			return 1;
+		}).subscribeOn(Schedulers.boundedElastic());
+
+		Mono<Integer> inflightMono2 = Mono.fromCallable(() -> {
+			try {
+				Thread.sleep(200);
+			} catch (InterruptedException e) {
+				counter.incrementAndGet();
+			}
+			return 2;
+		}).subscribeOn(Schedulers.boundedElastic());
+
+		Mono<Integer> completedMono = Mono.fromCallable(() -> {
+			Thread.sleep(100);
+			return 3;
+		}).subscribeOn(Schedulers.boundedElastic());
+
+		StepVerifier.withVirtualTime(() -> Mono.firstWithValue(inflightMono1, inflightMono2, completedMono))
+				.thenAwait(Duration.ofMillis(100))
+				.expectNext(3)
+				.verifyComplete();
+
+		assertThat(counter).hasValue(2L);
+	}
 
 	@Test
 	void singleArrayNullSource() {