ResourceCounter { public <K> Map<K, Metric> computeAggregates( Query.Builder query, Predicate<ITaskConfig> filter, Function<ITaskConfig, K> keyFunction) throws StorageException { LoadingCache<K, Metric> metrics = CacheBuilder.newBuilder() .build(new CacheLoader<K, Metric>() { @Override public Metric load(K key) { return new Metric(); } }); for (ITaskConfig task : Iterables.filter(getTasks(query), filter)) { metrics.getUnchecked(keyFunction.apply(task)).accumulate(task); } return metrics.asMap(); } @Inject  ResourceCounter(Storage storage); List<GlobalMetric> computeConsumptionTotals(); Metric computeQuotaAllocationTotals(); Map<K, Metric> computeAggregates(
      Query.Builder query,
      Predicate<ITaskConfig> filter,
      Function<ITaskConfig, K> keyFunction);  }
@Test public void testComputeAggregates() { insertTasks( task("bob", "jobA", "a", 1, GB, GB, PRODUCTION, RUNNING, NOT_DEDICATED), task("bob", "jobB", "b", 1, GB, GB, PRODUCTION, FAILED, NOT_DEDICATED), task("bob", "jobB", "b2", 1, GB, GB, PRODUCTION, FAILED, NOT_DEDICATED), task("bob", "jobC", "c", 1, GB, GB, NONPRODUCTION, RUNNING, NOT_DEDICATED), task("tim", "jobD", "d", 1, GB, GB, PRODUCTION, RUNNING, NOT_DEDICATED), task("bob", "jobE", "e", 1, GB, GB, NONPRODUCTION, ASSIGNED, NOT_DEDICATED), task("lil", "jobF", "f", 1, GB, GB, PRODUCTION, RUNNING, Optional.of("queue")), task("lil", "jobG", "g", 1, GB, GB, PRODUCTION, FINISHED, NOT_DEDICATED) ); assertEquals( ImmutableMap.of( JobKeys.from("bob", "test", "jobA"), new Metric(1, 1 * GB, 1 * GB), JobKeys.from("bob", "test", "jobB"), new Metric(2, 2 * GB, 2 * GB) ), resourceCounter.computeAggregates( Query.roleScoped("bob"), ITaskConfig::isProduction, ITaskConfig::getJob) ); }