@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_distributesFairlyWhenNotEven() {
    long totalItemsAndBytes = 10L;
    List<GetWorkBudgetSpender> streams = new ArrayList<>();
    for (int i = 0; i < 3; i++) {
      streams.add(spy(createGetWorkBudgetOwnerWithRemainingBudgetOf(GetWorkBudget.noBudget())));
    }
    createBudgetDistributor(0L)
        .distributeBudget(
            ImmutableList.copyOf(streams),
            GetWorkBudget.builder()
                .setItems(totalItemsAndBytes)
                .setBytes(totalItemsAndBytes)
                .build());

    long itemsAndBytesPerStream = (long) Math.ceil(totalItemsAndBytes / (streams.size() * 1.0));
    streams.forEach(
        stream ->
            verify(stream, times(1))
                .adjustBudget(eq(itemsAndBytesPerStream), eq(itemsAndBytesPerStream)));
  }