CF V3 getProcessStatistics - The threads are not released

Start with the applicationsV3().getProcessStatistics entry point and review issue 1054 alongside the reported requestAllApplicationStatistics usage. No repository file or test is identified in the issue, so first locate the implementation and existing tests, then reproduce the repeated calls. Done should be demonstrated by a regression test showing that threads do not keep increasing after statistics requests complete.

@twoseat I am tagging you directly as the issue still exists.
we have about 70 microservices running in the cloud foundry & we would like to monitor all the microservices with the following metrics,

• State
• Memory utilization
• CPU Utilization
• Disk Utilization

recently, I started to use CF v3 API & the issue is still the same. The allocated threads are not released, Due to which the application crashes multiple times a day.

The below API is called once every 5 minutes for about 70+ microservices

    public void requestAllApplicationStatistics(DefaultCloudFoundryOperations cloudFoundryOperations, List<CfArtifacts> cfArtifacts) {
        initializeTaskExecutor();

        Flux.fromIterable(cfArtifacts)
                .flatMap(application -> cloudFoundryOperations.getCloudFoundryClient().applicationsV3()
                        .getProcessStatistics(GetApplicationProcessStatisticsRequest.builder()
                                .applicationId(application.getId())
                                .type("web")
                                .build())
                        .map(response -> {
                            logger.debug("step 1 - cfStatistics fetched successfully");
                            CfApplicationResponse cfApplicationResponse = new CfApplicationResponse();
                            cfApplicationResponse.setApplicationId(application.getName());
                            cfApplicationResponse.setCfApplicationId(application.getId());
                            cfApplicationResponse.setPrefix(application.getPrefix());
                            List<ProcessStatisticsResource> resourceList = response.getResources();
                            cfApplicationResponse.setApplicationInstances(resourceList);
                            logger.debug("step 2 - cfStatistics before calling the callback");
                            return cfApplicationResponse;
                        })
                        .doOnError(e -> {
                            logger.error("cfStatistics Fetch Failed with error: {}", e.getMessage(), e);
                        })
                        .doFinally(signalType -> {
                            logger.debug("All applications processed. Shutting down the executor service.");
//                            shutdownTaskExecutor(); // Shutdown the executor service
                        })
                )
                .subscribeOn(Schedulers.fromExecutor(taskExecutor))// Use the shared task executor
                .publishOn(Schedulers.fromExecutor(taskExecutor)) // Use the shared task executor
                .doOnNext(cfResponseCallback::processCfMetrics)
                .subscribe(); // Subscribe to start the execution
    }

    private static synchronized void initializeTaskExecutor() {
        if (taskExecutor == null) {
            taskExecutor = new ThreadPoolTaskExecutor();
            taskExecutor.setCorePoolSize(10); // Set the desired core pool size
            taskExecutor.setMaxPoolSize(20); // Set the desired maximum pool size
            taskExecutor.setThreadNamePrefix("cf-task-executor-"); // Set a thread name prefix
            taskExecutor.initialize(); // Initialize the task executor
        }
    }

Look at the active threads in the below screenshot, it keeps increasing & crashes (the screenshot was captured from the Health Monitoring on the total threads measured by java Open telemetry instrumentation)