CF V3 getProcessStatistics - The threads are not released

Bắt đầu với điểm vào applicationsV3().getProcessStatistics và xem xét issue 1054 cùng với việc sử dụng requestAllApplicationStatistics đã được báo cáo. Vì issue không xác định tệp nào trong repository hoặc test nào, trước tiên hãy tìm implementation và các test hiện có, sau đó tái hiện các lần gọi lặp lại. Việc hoàn thành cần được chứng minh bằng một regression test cho thấy số lượng thread không tiếp tục tăng sau khi các yêu cầu thống kê hoàn tất.

@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)