class ClusterRemovableInstanceGroupAgentRemover {

    @VisibleForTesting
    Completable doRemoveAgents() {
        return Completable.defer(() -> {
            if (!configuration.isRemovingAgentsEnabled()) {
                logger.debug("Removing agents is not enabled");
                return Completable.complete();
            }

            long now = titusRuntime.getClock().wallTime();
            List<AgentInstanceGroup> eligibleInstanceGroups = agentManagementService.getInstanceGroups().stream()
                    .filter(agentManagementService::isOwnedByFenzo)
                    .filter(ig -> ig.getLifecycleStatus().getState() == InstanceGroupLifecycleState.Removable)
                    .filter(ig -> now - ig.getLifecycleStatus().getTimestamp() >= configuration.getInstanceGroupRemovableGracePeriodMs())
                    .collect(Collectors.toList());

            logger.debug("Eligible instance groups: {}", eligibleInstanceGroups);
            long totalAgentsToRemove = 0;
            Map<AgentInstanceGroup, List<AgentInstance>> agentInstancesPerInstanceGroup = new HashMap<>();
            for (AgentInstanceGroup instanceGroup : eligibleInstanceGroups) {
                List<AgentInstance> agentInstances = agentManagementService.getAgentInstances(instanceGroup.getId());
                totalAgentsToRemove += agentInstances.size();
                agentInstancesPerInstanceGroup.put(instanceGroup, agentInstances);
            }
            totalAgentsToRemoveGauge.set(totalAgentsToRemove);
            logger.debug("Agent instances per instance group: {}", agentInstancesPerInstanceGroup);

            if (totalAgentsToRemove <= 0) {
                totalEligibleAgentsToRemoveGauge.set(0);
                return Completable.complete();
            }

            long totalEligibleAgentsToRemove = 0;
            Map<AgentInstanceGroup, List<AgentInstance>> eligibleAgentInstancesPerInstanceGroup = new HashMap<>();
            Map<String, Long> numberOfTasksOnAgents = getNumberOfTasksOnAgents();
            logger.debug("numberOfTasksOnAgents: {}", numberOfTasksOnAgents);

            for (Map.Entry<AgentInstanceGroup, List<AgentInstance>> entry : agentInstancesPerInstanceGroup.entrySet()) {
                List<AgentInstance> eligibleAgentInstances = entry.getValue().stream()
                        .filter(i -> agentInstancesBeingTerminated.getIfPresent(i.getId()) == null)
                        .filter(i -> numberOfTasksOnAgents.getOrDefault(i.getId(), 0L) <= 0)
                        .collect(Collectors.toList());
                totalEligibleAgentsToRemove += eligibleAgentInstances.size();
                eligibleAgentInstancesPerInstanceGroup.put(entry.getKey(), eligibleAgentInstances);
            }
            totalEligibleAgentsToRemoveGauge.set(totalEligibleAgentsToRemove);
            logger.debug("Eligible agent instances per instance group: {}", eligibleAgentInstancesPerInstanceGroup);

            if (totalEligibleAgentsToRemove <= 0) {
                return Completable.complete();
            }

            long maxTokensToTake = Math.min(TOKEN_BUCKET_CAPACITY, totalEligibleAgentsToRemove);
            Optional<Pair<Long, ImmutableTokenBucket>> takeOpt = this.lastTokenBucket.tryTake(1, maxTokensToTake);
            if (takeOpt.isPresent()) {
                Pair<Long, ImmutableTokenBucket> takePair = takeOpt.get();
                this.lastTokenBucket = takePair.getRight();
                long tokensAvailable = takePair.getLeft();
                long tokensUsed = 0;

                logger.debug("Attempting to terminate {} agent instances", tokensAvailable);
                List<Completable> actions = new ArrayList<>();
                for (Map.Entry<AgentInstanceGroup, List<AgentInstance>> entry : eligibleAgentInstancesPerInstanceGroup.entrySet()) {
                    long tokensRemaining = tokensAvailable - tokensUsed;
                    if (tokensRemaining <= 0) {
                        break;
                    }
                    List<AgentInstance> agentInstances = entry.getValue();
                    if (agentInstances.isEmpty()) {
                        continue;
                    }
                    AgentInstanceGroup instanceGroup = entry.getKey();
                    String instanceGroupId = instanceGroup.getId();
                    int agentCountEligibleToTerminateInInstanceGroup = instanceGroup.getCurrent() - instanceGroup.getMin();
                    int agentCountToTerminate = Math.min((int) tokensRemaining, agentCountEligibleToTerminateInInstanceGroup);
                    if (agentCountToTerminate > 0) {
                        List<String> agentInstanceIdsToTerminate = agentInstances.stream()
                                .limit(agentCountToTerminate)
                                .map(AgentInstance::getId)
                                .collect(Collectors.toList());
                        logger.info("Terminating in instance group: {} agent instances({}): {}", instanceGroupId, agentInstanceIdsToTerminate.size(), agentInstanceIdsToTerminate);
                        actions.add(createTerminateAgentsCompletable(instanceGroupId, agentInstanceIdsToTerminate));
                        tokensUsed += agentInstanceIdsToTerminate.size();
                    }
                }
                totalAgentsBeingRemovedGauge.set(tokensUsed);
                return Completable.concat(actions);
            }
            return Completable.complete();
        }).doOnCompleted(() -> logger.debug("Completed cluster agent removal"))
                .timeout(REMOVE_AGENTS_COMPLETABLE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
    }

    @Inject public  ClusterRemovableInstanceGroupAgentRemover(TitusRuntime titusRuntime,
                                                     ClusterOperationsConfiguration configuration,
                                                     AgentManagementService agentManagementService,
                                                     V3JobOperations v3JobOperations);
    public  ClusterRemovableInstanceGroupAgentRemover(TitusRuntime titusRuntime,
                                                     ClusterOperationsConfiguration configuration,
                                                     AgentManagementService agentManagementService,
                                                     V3JobOperations v3JobOperations,
                                                     Scheduler scheduler);

}