rsocket/rsocket-java

Hi. Question about shutting down Spring context with Flux.interval() running.

I have the following code that pushes telemetry over RSocket at a given interval.

@Component
public class BackendService {
    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final String id = "test1";
    private final String backendUrl = "ws://localhost:9898";

    private final RSocketRequester requester;
    private final Disposable telemetrySubscriber;

    @PreDestroy
    public void destroy() {
        telemetrySubscriber.dispose();
    }

    public BackendService(RSocketRequester.Builder builder, RSocketStrategies strategies) {
        logger.info("id: {}, backend URL: {}", id, backendUrl);

        HttpClient httpClient = HttpClient.create()
                .baseUrl(backendUrl);

        requester = builder
                .setupRoute("/backend/connect")
                .setupData(id)
                .rsocketConnector(connector -> connector.reconnect(Retry.fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(1))))
                .dataMimeType(MediaType.APPLICATION_JSON)
                .transport(WebsocketClientTransport.create(httpClient, "/rsocket"));

        telemetrySubscriber = Flux.interval(Duration.ofSeconds(10))
            .onBackpressureDrop()
            .map(tick -> "Telemetry from " + id + " tick " + tick)
            .as(flux -> requester.route("/backend/telemetry").data(flux).retrieveFlux(Void.class))
            .subscribe();

    }

}

It works fine in general, but I have noticed that when I shut down Spring Context where this Component is defined, it doesn't shut down completely - there is a bunch of reactor threads that are still running (e.g. reactor-http-nio-X etc) and it seems that Flux.interval() is still active. The following exception is being printed at the given interval:

java.util.concurrent.RejectedExecutionException: event executor terminated
    at io.netty.util.concurrent.SingleThreadEventExecutor.reject(SingleThreadEventExecutor.java:926) ~[netty-common-4.1.58.Final.jar:4.1.58.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor.offerTask(SingleThreadEventExecutor.java:353) ~[netty-common-4.1.58.Final.jar:4.1.58.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor.addTask(SingleThreadEventExecutor.java:346) ~[netty-common-4.1.58.Final.jar:4.1.58.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor.execute(SingleThreadEventExecutor.java:828) ~[netty-common-4.1.58.Final.jar:4.1.58.Final]

The complete code is here: https://github.com/maximdim/rsocket2/blob/lifecycle/src/main/java/rsocket/client/BackendService.java

I would appreciate any hints. Thanks!