@Override
    public ResourceReconcileResult tryReconcileClusterResources(
            TaskManagerResourceInfoProvider taskManagerResourceInfoProvider) {
        ResourceReconcileResult.Builder builder = ResourceReconcileResult.builder();

        List<TaskManagerInfo> taskManagersIdleTimeout = new ArrayList<>();
        List<TaskManagerInfo> taskManagersNonTimeout = new ArrayList<>();
        long currentTime = System.currentTimeMillis();
        taskManagerResourceInfoProvider
                .getRegisteredTaskManagers()
                .forEach(
                        taskManagerInfo -> {
                            if (taskManagerInfo.isIdle()
                                    && currentTime - taskManagerInfo.getIdleSince()
                                            >= taskManagerTimeout.toMilliseconds()) {
                                taskManagersIdleTimeout.add(taskManagerInfo);
                            } else {
                                taskManagersNonTimeout.add(taskManagerInfo);
                            }
                        });

        List<PendingTaskManager> pendingTaskManagersNonUse = new ArrayList<>();
        List<PendingTaskManager> pendingTaskManagersInuse = new ArrayList<>();
        taskManagerResourceInfoProvider
                .getPendingTaskManagers()
                .forEach(
                        pendingTaskManager -> {
                            if (pendingTaskManager.getPendingSlotAllocationRecords().isEmpty()) {
                                pendingTaskManagersNonUse.add(pendingTaskManager);
                            } else {
                                pendingTaskManagersInuse.add(pendingTaskManager);
                            }
                        });

        ResourceProfile resourcesToKeep = ResourceProfile.ZERO;
        ResourceProfile resourcesInTotal = ResourceProfile.ZERO;
        boolean resourceFulfilled = false;

        // check whether available resources of used (pending) task manager is enough.
        ResourceProfile resourcesAvailableOfNonIdle =
                getAvailableResourceOfTaskManagers(taskManagersNonTimeout);

        ResourceProfile resourcesInTotalOfNonIdle =
                getTotalResourceOfTaskManagers(taskManagersNonTimeout);

        resourcesToKeep = resourcesToKeep.merge(resourcesAvailableOfNonIdle);
        resourcesInTotal = resourcesInTotal.merge(resourcesInTotalOfNonIdle);

        if (isRequiredResourcesFulfilled(resourcesToKeep, resourcesInTotal)) {
            resourceFulfilled = true;
        } else {
            ResourceProfile resourcesAvailableOfNonIdlePendingTaskManager =
                    getAvailableResourceOfPendingTaskManagers(pendingTaskManagersInuse);
            ResourceProfile resourcesInTotalOfNonIdlePendingTaskManager =
                    getTotalResourceOfPendingTaskManagers(pendingTaskManagersInuse);

            resourcesToKeep = resourcesToKeep.merge(resourcesAvailableOfNonIdlePendingTaskManager);
            resourcesInTotal = resourcesInTotal.merge(resourcesInTotalOfNonIdlePendingTaskManager);
        }

        // try reserve or release unused (pending) task managers
        for (TaskManagerInfo taskManagerInfo : taskManagersIdleTimeout) {
            if (resourceFulfilled
                    || isRequiredResourcesFulfilled(resourcesToKeep, resourcesInTotal)) {
                resourceFulfilled = true;
                builder.addTaskManagerToRelease(taskManagerInfo);
            } else {
                resourcesToKeep = resourcesToKeep.merge(taskManagerInfo.getAvailableResource());
                resourcesInTotal = resourcesInTotal.merge(taskManagerInfo.getTotalResource());
            }
        }
        for (PendingTaskManager pendingTaskManager : pendingTaskManagersNonUse) {
            if (resourceFulfilled
                    || isRequiredResourcesFulfilled(resourcesToKeep, resourcesInTotal)) {
                resourceFulfilled = true;
                builder.addPendingTaskManagerToRelease(pendingTaskManager);
            } else {
                resourcesToKeep = resourcesToKeep.merge(pendingTaskManager.getUnusedResource());
                resourcesInTotal =
                        resourcesInTotal.merge(pendingTaskManager.getTotalResourceProfile());
            }
        }

        if (!resourceFulfilled) {
            // fulfill required resources
            tryFulFillRequiredResourcesWithAction(
                    resourcesToKeep, resourcesInTotal, builder::addPendingTaskManagerToAllocate);
        }

        return builder.build();
    }
@Test
    void testIdlePendingTaskManagerShouldBeReleased() {
        final PendingTaskManager pendingTaskManager =
                new PendingTaskManager(DEFAULT_SLOT_RESOURCE, 1);
        final TaskManagerResourceInfoProvider taskManagerResourceInfoProvider =
                TestingTaskManagerResourceInfoProvider.newBuilder()
                        .setPendingTaskManagersSupplier(
                                () -> Collections.singleton(pendingTaskManager))
                        .build();

        ResourceReconcileResult result =
                ANY_MATCHING_STRATEGY.tryReconcileClusterResources(taskManagerResourceInfoProvider);

        assertThat(result.getPendingTaskManagersToRelease()).containsExactly(pendingTaskManager);
    }