Merge #2738 into 3.4.8

simonbasle · simonbasle · commit fd14969b5d59 · 2021-07-12T09:44:47.000+02:00
Added `FluxSwitchMapTest.test2596﻿()` as slow test, which is one
of the worst offenders.
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxBufferPredicateTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxBufferPredicateTest.java
@@ -36,6 +36,7 @@
 import java.util.stream.Collectors;
 
 import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
@@ -826,6 +827,7 @@ protected void hookOnSubscribe(Subscription subscription) {
 	}
 
 	@Test
+	@Tag("slow")
 	public void requestRaceWithOnNextLoops() {
 		for (int i = 0; i < ROUNDS; i++) {
 			requestRaceWithOnNext();
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxBufferWhenTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxBufferWhenTest.java
@@ -32,6 +32,7 @@
 import java.util.logging.Level;
 
 import org.assertj.core.api.Condition;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.reactivestreams.Subscription;
@@ -61,6 +62,7 @@ public class FluxBufferWhenTest {
 
 	//see https://github.com/reactor/reactor-core/issues/969
 	@Test
+	@Tag("slow")
 	public void bufferedCanCompleteIfOpenNeverCompletesDropping() {
 		//this test ensures that dropping buffers will complete if the source is exhausted before the open publisher finishes
 		Mono<Integer> buffered = Flux.range(1, 200)
@@ -82,6 +84,7 @@ public void bufferedCanCompleteIfOpenNeverCompletesDropping() {
 
 	//see https://github.com/reactor/reactor-core/issues/969
 	@Test
+	@Tag("slow")
 	public void bufferedCanCompleteIfOpenNeverCompletesOverlapping() {
 		//this test ensures that overlapping buffers will complete if the source is exhausted before the open publisher finishes
 		Mono<Integer> buffered = Flux.range(1, 200)
@@ -102,6 +105,7 @@ public void bufferedCanCompleteIfOpenNeverCompletesOverlapping() {
 
 	//see https://github.com/reactor/reactor-core/issues/969
 	@Test
+	@Tag("slow")
 	public void timedOutBuffersDontLeak() throws InterruptedException {
 		LongAdder created = new LongAdder();
 		MemoryUtils.RetainedDetector retainedDetector = new MemoryUtils.RetainedDetector();
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxGroupByTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxGroupByTest.java
@@ -23,6 +23,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.reactivestreams.Subscription;
@@ -387,6 +388,7 @@ public void twoGroupsLongAsyncMergeHidden() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void twoGroupsLongAsyncMergeHidden2() {
 		ForkJoinPool forkJoinPool = new ForkJoinPool();
 
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxRefCountGraceTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxRefCountGraceTest.java
@@ -22,6 +22,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
@@ -271,6 +272,7 @@ public void raceSubscribeAndCancel() {
 
 	//see https://github.com/reactor/reactor-core/issues/1260
 	@Test
+	@Tag("slow")
 	public void raceSubscribeAndCancelNoTimeout() {
 		final Flux<String> testFlux = Flux.<String>create(fluxSink -> fluxSink.next("Test").complete())
 				.replay(1)
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxSwitchMapTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxSwitchMapTest.java
@@ -21,6 +21,7 @@
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -43,6 +44,7 @@ public class FluxSwitchMapTest {
 	@ParameterizedTest
 	@ValueSource(ints = {0}) // TODO: add for prefetch one
 	@Timeout(120_000)
+	@Tag("slow")
 	// test for issue https://github.com/reactor/reactor-core/issues/2554
 	public void test2596() {
 		final Scheduler scheduler = Schedulers.newSingle("test");
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxTakeTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxTakeTest.java
@@ -22,6 +22,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 import org.reactivestreams.Publisher;
@@ -705,6 +706,7 @@ public void scanFuseableSubscriber() {
     }
 
     @Test
+	@Tag("slow")
 	public void onSubscribeRaceRequestingShouldBeConsistentForTakeFuseableTest() throws InterruptedException {
 		for (int i = 0; i < 5; i++) {
 			int take = 3000;
@@ -718,6 +720,7 @@ public void onSubscribeRaceRequestingShouldBeConsistentForTakeFuseableTest() thr
     }
 
 	@Test
+	@Tag("slow")
 	public void onSubscribeRaceRequestingShouldBeConsistentForTakeConditionalTest() throws InterruptedException {
 		for (int i = 0; i < 5; i++) {
 			int take = 3000;
@@ -732,6 +735,7 @@ public void onSubscribeRaceRequestingShouldBeConsistentForTakeConditionalTest()
 	}
 
 	@Test
+	@Tag("slow")
 	public void onSubscribeRaceRequestingShouldBeConsistentForTakeTest() throws InterruptedException {
 		for (int i = 0; i < 5; i++) {
 			int take = 3000;
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxWindowWhenTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxWindowWhenTest.java
@@ -26,6 +26,7 @@
 
 import org.assertj.core.api.Assertions;
 import org.assertj.core.api.Condition;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
@@ -65,6 +66,7 @@ static <T> void expect(AssertSubscriber<Flux<T>> ts, int index, T... values) {
 
 	//see https://github.com/reactor/reactor-core/issues/975
 	@Test
+	@Tag("slow")
 	public void noWindowRetained_gh975() throws InterruptedException {
 		LongAdder created = new LongAdder();
 		class Wrapper {
diff --git a/reactor-core/src/test/java/reactor/core/publisher/MonoCollectListTest.java b/reactor-core/src/test/java/reactor/core/publisher/MonoCollectListTest.java
@@ -26,6 +26,7 @@
 import java.util.function.Function;
 
 import org.assertj.core.api.SoftAssertions;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
@@ -43,6 +44,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static reactor.test.publisher.TestPublisher.Violation.CLEANUP_ON_TERMINATE;
 
+@Tag("slow")
 public class MonoCollectListTest {
 
 	static final Logger LOGGER = Loggers.getLogger(MonoCollectListTest.class);
@@ -218,6 +220,7 @@ public void discardOnCancel() {
 
 
 	@Test
+	@Tag("slow")
 	public void discardCancelNextRace() {
 		AtomicInteger doubleDiscardCounter = new AtomicInteger();
 		Context discardingContext = Operators.enableOnDiscard(null, o -> {
@@ -247,6 +250,7 @@ public void discardCancelNextRace() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void discardCancelCompleteRace() {
 		AtomicInteger doubleDiscardCounter = new AtomicInteger();
 		Context discardingContext = Operators.enableOnDiscard(null, o -> {
diff --git a/reactor-core/src/test/java/reactor/core/publisher/MonoCompletionStageTest.java b/reactor-core/src/test/java/reactor/core/publisher/MonoCompletionStageTest.java
@@ -21,6 +21,7 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Subscription;
 
@@ -74,6 +75,7 @@ public void cancelFutureImmediatelyCancelledLoop() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void cancelFutureDelayedCancelledLoop() {
 		for (int i = 0; i < 500; i++) {
 			CompletableFuture<Integer> future = new CompletableFuture<>();
@@ -95,6 +97,7 @@ public void cancelFutureDelayedCancelledLoop() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void cancelFutureTimeoutCancelledLoop() {
 		for (int i = 0; i < 500; i++) {
 			CompletableFuture<Integer> future = new CompletableFuture<>();
diff --git a/reactor-core/src/test/java/reactor/core/publisher/MonoIgnoreThenTest.java b/reactor-core/src/test/java/reactor/core/publisher/MonoIgnoreThenTest.java
@@ -19,6 +19,7 @@
 import java.time.Duration;
 
 import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
 
@@ -67,7 +68,8 @@ void justThenIgnoreWithCancel() {
 	}
 
 	@Test
-		// https://github.com/reactor/reactor-core/issues/2561
+	@Tag("slow")
+	// https://github.com/reactor/reactor-core/issues/2561
 	void raceTest2561() {
 		final Scheduler scheduler = Schedulers.newSingle("non-test-thread");
 		final Mono<String> getCurrentThreadName =
diff --git a/reactor-core/src/test/java/reactor/core/publisher/OnDiscardShouldNotLeakTest.java b/reactor-core/src/test/java/reactor/core/publisher/OnDiscardShouldNotLeakTest.java
@@ -28,6 +28,7 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.reactivestreams.Publisher;
@@ -44,10 +45,11 @@
 import reactor.util.annotation.Nullable;
 import reactor.util.concurrent.Queues;
 
-// TODO Junit 5: would maybe be better handled as a dynamic test, but  was migrated kind of "as is" to make sure
-// test count did not regress.
 import static reactor.core.publisher.Sinks.EmitFailureHandler.FAIL_FAST;
 
+// TODO Junit 5: would maybe be better handled as a dynamic test, but  was migrated kind of "as is" to make sure
+// test count did not regress.
+@Tag("slow")
 public class OnDiscardShouldNotLeakTest {
 
 	private static final int NB_ITERATIONS = 100;
diff --git a/reactor-core/src/test/java/reactor/core/publisher/ParallelFluxTest.java b/reactor-core/src/test/java/reactor/core/publisher/ParallelFluxTest.java
@@ -35,6 +35,7 @@
 import java.util.concurrent.atomic.LongAdder;
 
 import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.reactivestreams.Publisher;
@@ -62,6 +63,7 @@ public class ParallelFluxTest {
 	public AutoDisposingExtension afterTest = new AutoDisposingExtension();
 
 	@Test
+	@Tag("slow")
 	public void sequentialMode() {
 		Flux<Integer> source = Flux.range(1, 1_000_000)
 		                           .hide();
@@ -83,6 +85,7 @@ public void sequentialMode() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void sequentialModeFused() {
 		Flux<Integer> source = Flux.range(1, 1_000_000);
 		for (int i = 1; i < 33; i++) {
@@ -103,6 +106,7 @@ public void sequentialModeFused() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void parallelMode() {
 		Flux<Integer> source = Flux.range(1, 1_000_000)
 		                           .hide();
@@ -138,6 +142,7 @@ public void parallelMode() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void parallelModeFused() {
 		Flux<Integer> source = Flux.range(1, 1_000_000);
 		int ncpu = Math.max(8,
diff --git a/reactor-core/src/test/java/reactor/core/publisher/QueueDrainSubscriberTest.java b/reactor-core/src/test/java/reactor/core/publisher/QueueDrainSubscriberTest.java
@@ -18,6 +18,7 @@
 
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.openjdk.jol.info.ClassLayout;
 import org.openjdk.jol.info.FieldLayout;
@@ -27,6 +28,7 @@
 public class QueueDrainSubscriberTest {
 
 	@Test
+	@Tag("slow")
 	public void objectPadding() {
 		ClassLayout layout = ClassLayout.parseClass(QueueDrainSubscriber.class);
 		AtomicReference<FieldLayout> wip = new AtomicReference<>();
diff --git a/reactor-core/src/test/java/reactor/core/publisher/SerializedSubscriberTest.java b/reactor-core/src/test/java/reactor/core/publisher/SerializedSubscriberTest.java
@@ -21,6 +21,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Subscription;
 
@@ -38,6 +39,7 @@ public class SerializedSubscriberTest {
 
 	//see https://github.com/reactor/reactor-core/issues/2077
 	@Test
+	@Tag("slow")
 	public void onNextRaceWithCancelDoesNotLeak() {
 		int loops = 0;
 		while (loops < 100_000) {
diff --git a/reactor-core/src/test/java/reactor/core/publisher/scenarios/FluxTests.java b/reactor-core/src/test/java/reactor/core/publisher/scenarios/FluxTests.java
@@ -47,6 +47,7 @@
 import java.util.stream.IntStream;
 
 import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.api.extension.RegisterExtension;
@@ -552,6 +553,7 @@ public void analyticsTest() throws Exception {
 	}
 
 	@Test
+	@Tag("slow")
 	public void parallelTests() throws InterruptedException {
 		parallelMapManyTest("sync", 1_000_000);
 		parallelMapManyTest("shared", 1_000_000);
diff --git a/reactor-core/src/test/java/reactor/core/scheduler/BoundedElasticSchedulerTest.java b/reactor-core/src/test/java/reactor/core/scheduler/BoundedElasticSchedulerTest.java
@@ -48,6 +48,7 @@
 import org.awaitility.Awaitility;
 
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import reactor.core.Disposable;
 import reactor.core.Disposables;
@@ -171,6 +172,7 @@ public void doubleSubscribeOn() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void testLargeNumberOfWorkers() throws InterruptedException {
 		final int maxThreads = 3;
 		final int maxQueue = 10;
@@ -421,6 +423,7 @@ public void evictionForWorkerScheduling() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void lifoEvictionNoThreadRegrowth() throws InterruptedException {
 		int otherThreads = Thread.activeCount(); //don't count the evictor at shutdown
 		Set<String> preExistingEvictors = dumpThreadNames().filter(s -> s.startsWith("boundedElastic-evictor")).collect(Collectors.toSet());
@@ -1124,6 +1127,7 @@ public void publishOnDisposesWorkerWhenCancelled() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void pickSetIdleRaceBusy() {
 		BoundedElasticScheduler scheduler = afterTest.autoDispose(new BoundedElasticScheduler(1, 1, r -> new Thread(r, "test"),
 				1000));
@@ -1143,6 +1147,7 @@ public void pickSetIdleRaceBusy() {
 
 	}
 	@Test
+	@Tag("slow")
 	public void pickSetIdleRaceIdle() {
 		BoundedElasticScheduler scheduler = afterTest.autoDispose(new BoundedElasticScheduler(1, 1, r -> new Thread(r, "test"),
 				1000));
@@ -1178,6 +1183,7 @@ public void gh1992() {
 
 	//gh-1973 smoke test
 	@Test
+	@Tag("slow")
 	public void testGh1973() throws InterruptedException {
 		Scheduler scheduler = afterTest.autoDispose(Schedulers.newBoundedElastic(3, 100000, "subscriberElastic", 600, true));
 		LinkedList<MonoSink<String>> listeners = new LinkedList<>();
diff --git a/reactor-core/src/test/java/reactor/core/scheduler/SchedulersHooksTest.java b/reactor-core/src/test/java/reactor/core/scheduler/SchedulersHooksTest.java
@@ -24,6 +24,7 @@
 import java.util.function.Function;
 
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import reactor.test.AutoDisposingExtension;
@@ -42,6 +43,7 @@ public void resetAllHooks() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void onScheduleIsAdditive() throws Exception {
 		AtomicInteger tracker = new AtomicInteger();
 		Schedulers.onScheduleHook("k1", new TrackingDecorator(tracker, 1));
@@ -92,6 +94,7 @@ public void onScheduleIgnoresUnknownRemovals() {
 	}
 
 	@Test
+	@Tag("slow")
 	public void onScheduleResetOne() throws InterruptedException {
 		AtomicInteger tracker = new AtomicInteger();
 		Schedulers.onScheduleHook("k1", new TrackingDecorator(tracker, 1));
diff --git a/reactor-core/src/test/java/reactor/core/scheduler/SchedulersTest.java b/reactor-core/src/test/java/reactor/core/scheduler/SchedulersTest.java
