Fix memory leak in the FluxMessageChannel (#8622)

The `FluxMessageChannel` can subscribe to any volatile `Publisher`.
For example, we can call Reactor Kafka `Sender.send()` for
input data and pass its result to the `FluxMessageChannel`
for on demand subscription.
These publishers are subscribed in the `FluxMessageChannel`
and their `Disposable` is stored in the internal `Disposable.Composite`
which currently only cleared on `destroy()`

* Extract `Disposable` from those internal `subscribe()` calls
into an `AtomicReference`.
* Use this `AtomicReference` in the `doOnTerminate()`
to remove from the `Disposable.Composite` and `dispose()`
when such a volatile `Publisher` is completed

**Cherry-pick to `6.0.x` & `5.5.x`**

spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java

@@ -26,6 +26,8 @@ import java.util.stream.IntStream;
 import org.junit.jupiter.api.Test;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
@@ -144,6 +146,25 @@ public class FluxMessageChannelTests {
 				.until(() -> TestUtils.getPropertyValue(flux, "sink.sink.done", Boolean.class));
 	}
 
+	@Test
+	void noMemoryLeakInFluxMessageChannelForVolatilePublishers() {
+		FluxMessageChannel messageChannel = new FluxMessageChannel();
+
+		StepVerifier stepVerifier = StepVerifier.create(messageChannel)
+				.expectNextCount(3)
+				.thenCancel()
+				.verifyLater();
+
+		messageChannel.subscribeTo(Mono.just(new GenericMessage<>("test")));
+		messageChannel.subscribeTo(Flux.just("test1", "test2").map(GenericMessage::new));
+
+		stepVerifier.verify();
+
+		Disposable.Composite upstreamSubscriptions =
+				TestUtils.getPropertyValue(messageChannel, "upstreamSubscriptions", Disposable.Composite.class);
+		assertThat(upstreamSubscriptions.size()).isEqualTo(0);
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class TestConfiguration {