Make RetryWhenMainSubscriber#onError's calls serial (#2499, Fixes #2488)

Since `#onError` is called by multiple `Publisher`s, we cannot use `FAIL_FAST` emit strategy
and instead should retry non-serialized access.

Author: bsideup

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

@@ -186,9 +186,9 @@ public void onError(Throwable t) {
 				produced(p);
 			}
 
-			otherArbiter.request(1);
-
 			signaller.emitNext(this, Sinks.EmitFailureHandler.FAIL_FAST);
+			// request after signalling, otherwise it may race
+			otherArbiter.request(1);
 		}
 
 		@Override

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

@@ -1119,4 +1119,19 @@ public void retryWhenWithThrowableFunction() {
 		            .verifyComplete();
 	}
 
+	@Test
+	void gh2488() {
+		for (int i = 0; i < 1_000; i++) {
+			AtomicInteger sourceHelper = new AtomicInteger();
+			Flux.just("hello")
+			    .doOnNext(m -> {
+				    if (sourceHelper.getAndIncrement() < 9) {
+					    throw new RuntimeException("Boom!");
+				    }
+			    })
+			    .retryWhen(Retry.fixedDelay(10, Duration.ofNanos(1)))
+			    .blockFirst(Duration.ofSeconds(1));
+		}
+	}
+
 }