Repository: reactor/reactor-core
Issue: `FluxGroupBy`'s `GroupedFlux` does not propagate the `Subscription` to the second (rejected) `Subscriber`

When calling flatMap on a Publisher which throws an exception upon subscription, there are certain cases where checkTerminated throws a NPE and masks the actual exception. Furthermore flatMap never terminates.

## Expected Behavior
flatMap terminates with the actual Error

## Actual Behavior
NPE and no Termination. Stacktrace from the example below

```
2023-08-02 13:04:02  ERROR Operators - Operator called default onErrorDropped
java.lang.NullPointerException: null
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.checkTerminated(FluxFlatMap.java:840) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:609) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:589) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.onError(FluxFlatMap.java:452) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.subscribe(FluxGroupBy.java:721) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.InternalFluxOperator.subscribe(InternalFluxOperator.java:62) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxRetry$RetrySubscriber.resubscribe(FluxRetry.java:117) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxRetry.subscribeOrReturn(FluxRetry.java:52) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.Flux.subscribe(Flux.java:8759) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:427) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxGroupBy$GroupByMain.drainLoop(FluxGroupBy.java:393) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxGroupBy$GroupByMain.drain(FluxGroupBy.java:329) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:207) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxArray$ArraySubscription.slowPath(FluxArray.java:127) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxArray$ArraySubscription.request(FluxArray.java:100) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxGroupBy$GroupByMain.onSubscribe(FluxGroupBy.java:171) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:53) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:59) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.core.publisher.Flux.subscribe(Flux.java:8773) ~[reactor-core-3.5.8.jar:3.5.8]
	at reactor.test.DefaultStepVerifierBuilder$DefaultStepVerifier.toVerifierAndSubscribe(DefaultStepVerifierBuilder.java:891) ~[reactor-test-3.5.8.jar:3.5.8]
	at reactor.test.DefaultStepVerifierBuilder$DefaultStepVerifier.verify(DefaultStepVerifierBuilder.java:831) ~[reactor-test-3.5.8.jar:3.5.8]
	at reactor.test.DefaultStepVerifierBuilder$DefaultStepVerifier.verify(DefaultStepVerifierBuilder.java:823) ~[reactor-test-3.5.8.jar:3.5.8]
	at reactor.test.DefaultStepVerifierBuilder.verifyErrorSatisfies(DefaultStepVerifierBuilder.java:685) ~[reactor-test-3.5.8.jar:3.5.8]
```
## Steps to Reproduce
```kotlin
    @Test
    fun flatMapExceptionTest() {
        StepVerifier.create(Flux.just(1, 2, 3, 4, 5, 6)
            .groupBy {
                it > 3
            }
            .flatMap {
                it.flatMap<Int> { Mono.error(Error("whoopsies")) }.retry(2)
            }
        ).verifyErrorSatisfies {
            assertThat(it).hasMessage("whoopsies")
        }
    }
```
The actual error here (GroupedFlux allows only one Subscriber) is hidden by the NPE

## Possible Solution
Null check? Not sure if there is more behind it.

## Your Environment
* Reactor version(s) used: 3.5.8
* JVM version (`java -version`): openjdk version "11.0.17" 2022-10-18 LTS
* OS and version (eg `uname -a`): Linux S0155454 6.2.0-26-generic


Changed files:
reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java, reactor-core/src/test/java/reactor/core/publisher/FluxGroupByTest.java

diff --git a/reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java b/reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java
index 30cad47c3..08092c8e0 100644
--- a/reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.java
+++ b/reactor-core/src/main/java/reactor/core/publisher/FluxGroupBy.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.
@@ -718,7 +718,7 @@ final class FluxGroupBy<T, K, V> extends InternalFluxOperator<T, GroupedFlux<K,
 				drain();
 			}
 			else {
-				actual.onError(new IllegalStateException(
+				Operators.error(actual, new IllegalStateException(
 						"GroupedFlux allows only one Subscriber"));
 			}
 		}
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxGroupByTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxGroupByTest.java
index fdfeb524d..00d974d06 100644
--- a/reactor-core/src/test/java/reactor/core/publisher/FluxGroupByTest.java
+++ b/reactor-core/src/test/java/reactor/core/publisher/FluxGroupByTest.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.
@@ -47,6 +47,16 @@ import static org.assertj.core.api.Assertions.fail;
 public class FluxGroupByTest extends
                              FluxOperatorTest<String, GroupedFlux<Integer, String>> {
 
+	@Test
+	// see https://github.com/reactor/reactor-core/issues/3554
+	void flatMapExceptionTest() {
+		Flux.just(1, 2, 3, 4, 5, 6)
+		    .groupBy(it -> it > 3)
+		    .flatMap(it -> it.flatMap(__ -> Mono.error(new RuntimeException("whoopsies"))).retry(2))
+		    .as(StepVerifier::create)
+		    .verifyErrorSatisfies(it -> assertThat(it).hasMessage("GroupedFlux allows only one Subscriber"));
+	}
+
 	@Test
 	@Tag("slow")
 	//see https://github.com/reactor/reactor-core/issues/2730