#vulnerable code 
                              private Map<String, Resource> computeResourceBestPossibleStateWithWagedRebalancer(
      ResourceControllerDataProvider cache, CurrentStateOutput currentStateOutput,
      HelixManager helixManager, Map<String, Resource> resourceMap, BestPossibleStateOutput output,
      List<String> failureResources) {
    if (cache.isMaintenanceModeEnabled()) {
      // The WAGED rebalancer won't be used while maintenance mode is enabled.
      return Collections.emptyMap();
    }

    // Find the compatible resources: 1. FULL_AUTO 2. Configured to use the WAGED rebalancer
    Map<String, Resource> wagedRebalancedResourceMap =
        resourceMap.entrySet().stream().filter(resourceEntry -> {
          IdealState is = cache.getIdealState(resourceEntry.getKey());
          return is != null && is.getRebalanceMode().equals(IdealState.RebalanceMode.FULL_AUTO)
              && WagedRebalancer.class.getName().equals(is.getRebalancerClassName());
        }).collect(Collectors.toMap(resourceEntry -> resourceEntry.getKey(),
            resourceEntry -> resourceEntry.getValue()));

    Map<String, IdealState> newIdealStates = new HashMap<>();

    // Init rebalancer with the rebalance preferences.
    Map<ClusterConfig.GlobalRebalancePreferenceKey, Integer> preferences =
        cache.getClusterConfig().getGlobalRebalancePreference();

    if (METRIC_COLLECTOR_THREAD_LOCAL.get() == null) {
      try {
        // If HelixManager is null, we just pass in null for MetricCollector so that a
        // non-functioning WagedRebalancerMetricCollector would be created in WagedRebalancer's
        // constructor. This is to handle two cases: 1. HelixManager is null for non-testing cases -
        // in this case, WagedRebalancer will not read/write to metadata store and just use
        // CurrentState-based rebalancing. 2. Tests that require instrumenting the rebalancer for
        // verifying whether the cluster has converged.
        METRIC_COLLECTOR_THREAD_LOCAL.set(helixManager == null ? null
            : new WagedRebalancerMetricCollector(helixManager.getClusterName()));
      } catch (JMException e) {
        LogUtil.logWarn(logger, _eventId, String.format(
            "MetricCollector instantiation failed! WagedRebalancer will not emit metrics due to JMException %s",
            e));
      }
    }

    // TODO avoid creating the rebalancer on every rebalance call for performance enhancement
    WagedRebalancer wagedRebalancer =
        new WagedRebalancer(helixManager, preferences, METRIC_COLLECTOR_THREAD_LOCAL.get());
    try {
      newIdealStates.putAll(wagedRebalancer.computeNewIdealStates(cache, wagedRebalancedResourceMap,
          currentStateOutput));
    } catch (HelixRebalanceException ex) {
      // Note that unlike the legacy rebalancer, the WAGED rebalance won't return partial result.
      // Since it calculates for all the eligible resources globally, a partial result is invalid.
      // TODO propagate the rebalancer failure information to updateRebalanceStatus for monitoring.
      LogUtil.logError(logger, _eventId,
          String.format(
              "Failed to calculate the new Ideal States using the rebalancer %s due to %s",
              wagedRebalancer.getClass().getSimpleName(), ex.getFailureType()),
          ex);
    } finally {
      wagedRebalancer.close();
    }
    Iterator<Resource> itr = wagedRebalancedResourceMap.values().iterator();
    while (itr.hasNext()) {
      Resource resource = itr.next();
      IdealState is = newIdealStates.get(resource.getResourceName());
      // Check if the WAGED rebalancer has calculated the result for this resource or not.
      if (is != null && checkBestPossibleStateCalculation(is)) {
        // The WAGED rebalancer calculates a valid result, record in the output
        updateBestPossibleStateOutput(output, resource, is);
      } else {
        failureResources.add(resource.getResourceName());
        LogUtil.logWarn(logger, _eventId, String
            .format("Failed to calculate best possible states for %s.",
                resource.getResourceName()));
      }
    }
    return wagedRebalancedResourceMap;
  } 
                            #location 58 
                            #vulnerability type RESOURCE_LEAK
#fixed code
                              private Map<String, Resource> computeResourceBestPossibleStateWithWagedRebalancer(
      ResourceControllerDataProvider cache, CurrentStateOutput currentStateOutput,
      HelixManager helixManager, Map<String, Resource> resourceMap, BestPossibleStateOutput output,
      List<String> failureResources) {
    if (cache.isMaintenanceModeEnabled()) {
      // The WAGED rebalancer won't be used while maintenance mode is enabled.
      return Collections.emptyMap();
    }

    // Find the compatible resources: 1. FULL_AUTO 2. Configured to use the WAGED rebalancer
    Map<String, Resource> wagedRebalancedResourceMap =
        resourceMap.entrySet().stream().filter(resourceEntry -> {
          IdealState is = cache.getIdealState(resourceEntry.getKey());
          return is != null && is.getRebalanceMode().equals(IdealState.RebalanceMode.FULL_AUTO)
              && WagedRebalancer.class.getName().equals(is.getRebalancerClassName());
        }).collect(Collectors.toMap(resourceEntry -> resourceEntry.getKey(),
            resourceEntry -> resourceEntry.getValue()));

    Map<String, IdealState> newIdealStates = new HashMap<>();

    // Init rebalancer with the rebalance preferences.
    Map<ClusterConfig.GlobalRebalancePreferenceKey, Integer> preferences =
        cache.getClusterConfig().getGlobalRebalancePreference();

    // Create MetricCollector ThreadLocal if it hasn't been already initialized
    if (METRIC_COLLECTOR_THREAD_LOCAL.get() == null) {
      try {
        // If HelixManager is null, we just pass in null for MetricCollector so that a
        // non-functioning WagedRebalancerMetricCollector would be created in WagedRebalancer's
        // constructor. This is to handle two cases: 1. HelixManager is null for non-testing cases -
        // in this case, WagedRebalancer will not read/write to metadata store and just use
        // CurrentState-based rebalancing. 2. Tests that require instrumenting the rebalancer for
        // verifying whether the cluster has converged.
        METRIC_COLLECTOR_THREAD_LOCAL.set(helixManager == null ? null
            : new WagedRebalancerMetricCollector(helixManager.getClusterName()));
      } catch (JMException e) {
        LogUtil.logWarn(logger, _eventId, String.format(
            "MetricCollector instantiation failed! WagedRebalancer will not emit metrics due to JMException %s",
            e));
      }
    }

    // Create MetricCollector ThreadLocal if it hasn't been already initialized
    if (WAGED_REBALANCER_THREAD_LOCAL.get() == null) {
      WAGED_REBALANCER_THREAD_LOCAL
          .set(new WagedRebalancer(helixManager, preferences, METRIC_COLLECTOR_THREAD_LOCAL.get()));
    }
    WagedRebalancer wagedRebalancer = WAGED_REBALANCER_THREAD_LOCAL.get();

    try {
      newIdealStates.putAll(wagedRebalancer.computeNewIdealStates(cache, wagedRebalancedResourceMap,
          currentStateOutput));
    } catch (HelixRebalanceException ex) {
      // Note that unlike the legacy rebalancer, the WAGED rebalance won't return partial result.
      // Since it calculates for all the eligible resources globally, a partial result is invalid.
      // TODO propagate the rebalancer failure information to updateRebalanceStatus for monitoring.
      LogUtil.logError(logger, _eventId,
          String.format(
              "Failed to calculate the new Ideal States using the rebalancer %s due to %s",
              wagedRebalancer.getClass().getSimpleName(), ex.getFailureType()),
          ex);
    }

    Iterator<Resource> itr = wagedRebalancedResourceMap.values().iterator();
    while (itr.hasNext()) {
      Resource resource = itr.next();
      IdealState is = newIdealStates.get(resource.getResourceName());
      // Check if the WAGED rebalancer has calculated the result for this resource or not.
      if (is != null && checkBestPossibleStateCalculation(is)) {
        // The WAGED rebalancer calculates a valid result, record in the output
        updateBestPossibleStateOutput(output, resource, is);
      } else {
        failureResources.add(resource.getResourceName());
        LogUtil.logWarn(logger, _eventId, String
            .format("Failed to calculate best possible states for %s.",
                resource.getResourceName()));
      }
    }
    return wagedRebalancedResourceMap;
  }