Repository: reactor/reactor-core
Issue: FlatMapDelayError hangs on callable source error

<!--- Provide a general summary of the issue in the Title above -->
FlatMapDelayError does not propagate an onError signal when a callable source produces an error. 
It also does not cancel the upstream subscription. This results in a hanging flux that can never complete because it 
does not process any more values from the upstream publisher nor does it signal a termination to downstream subscribers.

Notably this only occurs when the error is from a callable source. When the mapping function throws the error directly 
then the behavior is as expected where the backlog is consumed and the flux terminates with an error. However when the 
operator is modified with onErrorConsume the error is properly swallowed and processing continues.

This does not affect concatMapDelayError as far as I have been able to test

<!--- /!\\ Make sure to follow the Contribution Guidelines, notably for security issues and questions:
https://github.com/reactor/.github/blob/main/CONTRIBUTING.md
https://pivotal.io/security
https://github.com/reactor/.github/blob/main/CONTRIBUTING.md#question-do-you-have-a-question
-->

## Expected Behavior
<!--- Tell us what you think should happen. -->
FlatMapDelayError should cancel the upstream subscription, process the backlog, then emit an onError response

## Actual Behavior
<!--- Tell us what happens instead of the expected behavior. -->
FlatMapDelayError never completes when callable source throws an error

## Possible Solution
<!--- Not obligatory, but you can suggest a fix/reason for the bug. -->
This behavior seems to have been introduced here 
https://github.com/reactor/reactor-core/commit/af0cb628ece3c0a543512ea9913c4ff72ba18566#diff-3ad99cc26a2bc1707dda2203da9a667ccf3abdd52aac45ee6b2c4cea438a842fR361-R363. 
However I do not know the side effects well enough to give a full solution.

Additionally it appears that this is caused by some optimization around the flatMap operator as there is a current test which utilizes .hide() on the error flux and completes successfully
linked here https://github.com/reactor/reactor-core/blob/912441f7d36adfb84f06595cecd8895b7518d8c9/reactor-core/src/test/java/reactor/core/publisher/FluxFlatMapTest.java#L604

A difference when using hide though is that the upstream subscription is still never cancelled so there is still a deviation of behavior from the thrown case as this would cause hot infinite publishers to never propagate the error and continue requesting events.


## Steps to Reproduce
<!---Provide a link to a live example, or an unambiguous set of steps to
reproduce this bug, eg. a unit test. Include code to reproduce, if relevant.
Most projects use JUnit5 now (like in snippet below; otherwise use JUnit4).
-->

Below are examples for the direct throw vs callable source

```java
@Test
void  workingFlatMapDelayError() {
	Flux.just(0, 1, 2, 3).log()
            .flatMapDelayError(integer -> {
                throw new RuntimeException(); // Cancels upstream subscription after consuming one event
            }, 1, 1)
            .as(StepVerifier::create)
            .expectError()
            .verify(Duration.ofSeconds(1)); // Completes as expected
}

@Test
void  hangingFlatMapDelayError() {
	Flux.just(0, 1, 2, 3).log()
            .flatMapDelayError(integer -> {
                return Flux.error(new RuntimeException()); // Does not cancel upstream subscription
            }, 1, 1)
            .as(StepVerifier::create)
            .expectError()
            .verify(Duration.ofSeconds(1)); // Triggers timeout
}

@Test
void  deoptimizedFlatMapDelayError() {
	Flux.just(0, 1, 2, 3).log()
            .flatMapDelayError(integer -> {
                return Flux.error(new RuntimeException()).hide(); // Does not cancel upstream subscription
            }, 1, 1)
            .as(StepVerifier::create)
            .expectError()
            .verify(Duration.ofSeconds(1)); // Completes after consuming all events
}
```

## Your Environment
<!--- Include as many relevant details about the environment you experienced the bug in. -->
<!--- Especially, always include the version(s) of Reactor library/libraries you used! -->

* Reactor version(s) used: reactor-core 3.5.2, reactor-test 3.5.2
* Other relevant libraries versions (eg. `netty`, ...): N/A
* JVM version (`java -version`): 17
* OS and version (eg `uname -a`): Windows 11

Changed files:
reactor-core/src/main/java/reactor/core/publisher/FluxFlatMap.java, reactor-core/src/test/java/reactor/core/publisher/FluxFlatMapTest.java

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxFlatMap.java b/reactor-core/src/main/java/reactor/core/publisher/FluxFlatMap.java
index f3959edd3..e14d6eb1f 100644
--- a/reactor-core/src/main/java/reactor/core/publisher/FluxFlatMap.java
+++ b/reactor-core/src/main/java/reactor/core/publisher/FluxFlatMap.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 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.
@@ -416,6 +416,7 @@ final class FluxFlatMap<T, R> extends InternalFluxOperator<T, R> {
 						onError(Operators.onOperatorError(s, e_, t, ctx));
 					}
 					Operators.onDiscard(t, ctx);
+					tryEmitScalar(null);
 					return;
 				}
 				tryEmitScalar(v);
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxFlatMapTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxFlatMapTest.java
index 7176da584..599772402 100644
--- a/reactor-core/src/test/java/reactor/core/publisher/FluxFlatMapTest.java
+++ b/reactor-core/src/test/java/reactor/core/publisher/FluxFlatMapTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 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.
@@ -42,7 +42,6 @@ import reactor.core.Scannable;
 import reactor.core.TestLoggerExtension;
 import reactor.core.publisher.FluxPeekFuseableTest.AssertQueueSubscription;
 import reactor.core.scheduler.Schedulers;
-import reactor.test.util.LoggerUtils;
 import reactor.test.StepVerifier;
 import reactor.test.publisher.TestPublisher;
 import reactor.test.subscriber.AssertSubscriber;
@@ -1869,4 +1868,41 @@ public class FluxFlatMapTest {
 		    .expectError(NoSuchMethodException.class)
 		    .verify();
 	}
+
+	static class FluxFlatMapDelayError3336Test {
+
+		@Test
+		void workingFlatMapDelayError() {
+			Flux.just(0, 1, 2, 3)
+			    .flatMapDelayError(integer -> {
+				    throw new RuntimeException(); // Cancels upstream subscription after consuming one event
+			    }, 1, 1)
+			    .as(StepVerifier::create)
+			    .expectError()
+			    .verify(Duration.ofSeconds(1)); // Completes as expected
+		}
+
+		@Test
+		void hangingFlatMapDelayError() {
+			Flux.just(0, 1, 2, 3)
+			    .flatMapDelayError(integer -> {
+				    return Flux.error(new RuntimeException()); // Does not cancel upstream subscription
+			    }, 1, 1)
+			    .as(StepVerifier::create)
+			    .expectError()
+			    .verify(Duration.ofSeconds(1)); // Triggers timeout
+		}
+
+		@Test
+		void deoptimizedFlatMapDelayError() {
+			Flux.just(0, 1, 2, 3)
+			    .flatMapDelayError(integer -> {
+				    return Flux.error(new RuntimeException())
+				               .hide(); // Does not cancel upstream subscription
+			    }, 1, 1)
+			    .as(StepVerifier::create)
+			    .expectError()
+			    .verify(Duration.ofSeconds(1)); // Completes after consuming all events
+		}
+	}
 }