@Override
  public <T extends GetWorkBudgetSpender> void distributeBudget(
      ImmutableCollection<T> budgetOwners, GetWorkBudget getWorkBudget) {
    if (budgetOwners.isEmpty()) {
      LOG.debug("Cannot distribute budget to no owners.");
      return;
    }

    if (getWorkBudget.equals(GetWorkBudget.noBudget())) {
      LOG.debug("Cannot distribute 0 budget.");
      return;
    }

    Map<T, GetWorkBudget> desiredBudgets = computeDesiredBudgets(budgetOwners, getWorkBudget);

    for (Entry<T, GetWorkBudget> streamAndDesiredBudget : desiredBudgets.entrySet()) {
      GetWorkBudgetSpender getWorkBudgetSpender = streamAndDesiredBudget.getKey();
      GetWorkBudget desired = streamAndDesiredBudget.getValue();
      GetWorkBudget remaining = getWorkBudgetSpender.remainingBudget();
      if (isBelowFiftyPercentOfTarget(remaining, desired)) {
        GetWorkBudget adjustment = desired.subtract(remaining);
        getWorkBudgetSpender.adjustBudget(adjustment);
      }
    }
  }
@Test
  public void testDistributeBudget_doesNothingWhenPassedInStreamsEmpty() {
    createBudgetDistributor(1L)
        .distributeBudget(
            ImmutableList.of(), GetWorkBudget.builder().setItems(10L).setBytes(10L).build());
  }