killbill-memoizeit

diff --git a/api/src/main/java/org/killbill/billing/junction/BillingEventSet.java b/api/src/main/java/org/killbill/billing/junction/BillingEventSet.java
index 8cffd89..432ecb5 100644
--- a/api/src/main/java/org/killbill/billing/junction/BillingEventSet.java
+++ b/api/src/main/java/org/killbill/billing/junction/BillingEventSet.java
@@ -17,16 +17,20 @@
 package org.killbill.billing.junction;
 
 import java.util.List;
+import java.util.Map;
 import java.util.SortedSet;
 import java.util.UUID;
 
 import org.killbill.billing.catalog.api.BillingMode;
+import org.killbill.billing.catalog.api.Usage;
 
 public interface BillingEventSet extends SortedSet<BillingEvent> {
 
-    public abstract boolean isAccountAutoInvoiceOff();
+    public boolean isAccountAutoInvoiceOff();
 
-    public abstract BillingMode getRecurringBillingMode();
+    public BillingMode getRecurringBillingMode();
 
-    public abstract List<UUID> getSubscriptionIdsWithAutoInvoiceOff();
+    public List<UUID> getSubscriptionIdsWithAutoInvoiceOff();
+
+    public Map<String, Usage> getUsages();
 }

diff --git a/api/src/main/java/org/killbill/billing/junction/BillingInternalApi.java b/api/src/main/java/org/killbill/billing/junction/BillingInternalApi.java
index 614b32f..f8cda97 100644
--- a/api/src/main/java/org/killbill/billing/junction/BillingInternalApi.java
+++ b/api/src/main/java/org/killbill/billing/junction/BillingInternalApi.java
@@ -18,6 +18,9 @@ package org.killbill.billing.junction;
 
 import java.util.UUID;
 
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.joda.time.LocalDate;
 import org.killbill.billing.callcontext.InternalCallContext;
 
 public interface BillingInternalApi {

diff --git a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/usage/TestConsumableInArrear.java b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/usage/TestConsumableInArrear.java
index 4cce354..b7ea0d7 100644
--- a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/usage/TestConsumableInArrear.java
+++ b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/usage/TestConsumableInArrear.java
@@ -73,7 +73,7 @@ public class TestConsumableInArrear extends TestIntegrationBase {
         //
         // CREATE SUBSCRIPTION AND EXPECT BOTH EVENTS: NextEvent.CREATE NextEvent.INVOICE
         //
-        final DefaultEntitlement bpSubscription = createBaseEntitlementAndCheckForCompletion(account.getId(), "bundleKey", "Shotgun", ProductCategory.BASE, BillingPeriod.MONTHLY, NextEvent.CREATE, NextEvent.INVOICE);
+        final DefaultEntitlement bpSubscription = createBaseEntitlementAndCheckForCompletion(account.getId(), "bundleKey", "Shotgun", ProductCategory.BASE, BillingPeriod.ANNUAL, NextEvent.CREATE, NextEvent.INVOICE);
         // Check bundle after BP got created otherwise we get an error from auditApi.
         subscriptionChecker.checkSubscriptionCreated(bpSubscription.getId(), internalCallContext);
         invoiceChecker.checkInvoice(account.getId(), 1, callContext, new ExpectedInvoiceItemCheck(new LocalDate(2012, 4, 1), null, InvoiceItemType.FIXED, new BigDecimal("0")));
@@ -82,28 +82,50 @@ public class TestConsumableInArrear extends TestIntegrationBase {
         //
         // ADD ADD_ON ON THE SAME DAY
         //
-        setUsage();
+        final List<RolledUpUsage> usageList = new ArrayList<RolledUpUsage>();
+        setUsage(usageList);
         addAOEntitlementAndCheckForCompletion(bpSubscription.getBundleId(), "Bullets", ProductCategory.ADD_ON, BillingPeriod.NO_BILLING_PERIOD, NextEvent.CREATE);
 
-        final RolledUpUsage usage = new DefaultRolledUpUsage(UUID.randomUUID(), "bullets", new LocalDate(2012, 4, 1).toDateTimeAtStartOfDay(DateTimeZone.UTC), new LocalDate(2012, 5, 1).toDateTimeAtStartOfDay(DateTimeZone.UTC), new BigDecimal("199"));
-        setUsage(usage);
+        RolledUpUsage usage1 = new DefaultRolledUpUsage(UUID.randomUUID(), "bullets", new LocalDate(2012, 4, 1).toDateTimeAtStartOfDay(DateTimeZone.UTC), new LocalDate(2012, 5, 1).toDateTimeAtStartOfDay(DateTimeZone.UTC), new BigDecimal("199"));
+        usageList.add(usage1);
+        setUsage(usageList);
 
         busHandler.pushExpectedEvents(NextEvent.PHASE, NextEvent.INVOICE, NextEvent.PAYMENT);
         clock.setDay(new LocalDate(2012, 5, 1));
         assertListenerStatus();
 
         invoiceChecker.checkInvoice(account.getId(), 2, callContext,
-                                    new ExpectedInvoiceItemCheck(new LocalDate(2012, 5, 1), new LocalDate(2012, 6, 1), InvoiceItemType.RECURRING, new BigDecimal("249.95")),
+                                    new ExpectedInvoiceItemCheck(new LocalDate(2012, 5, 1), new LocalDate(2013, 5, 1), InvoiceItemType.RECURRING, new BigDecimal("2399.95")),
                                     new ExpectedInvoiceItemCheck(new LocalDate(2012, 4, 1), new LocalDate(2012, 5, 1), InvoiceItemType.USAGE, new BigDecimal("5.90")));
 
 
+        // No usage for that period, we should still be schedule
+        RolledUpUsage usage2 = new DefaultRolledUpUsage(UUID.randomUUID(), "bullets", new LocalDate(2012, 5, 1).toDateTimeAtStartOfDay(DateTimeZone.UTC), new LocalDate(2012, 6, 1).toDateTimeAtStartOfDay(DateTimeZone.UTC), new BigDecimal("0"));
+        usageList.add(usage2);
+        setUsage(usageList);
+
+        busHandler.pushExpectedEvents(NextEvent.INVOICE);
+        clock.setDay(new LocalDate(2012, 6, 1));
+        assertListenerStatus();
+
+        invoiceChecker.checkInvoice(account.getId(), 3, callContext,
+                                    new ExpectedInvoiceItemCheck(new LocalDate(2012, 5, 1), new LocalDate(2012, 6, 1), InvoiceItemType.USAGE, BigDecimal.ZERO));
+
+
+        RolledUpUsage usage3 = new DefaultRolledUpUsage(UUID.randomUUID(), "bullets", new LocalDate(2012, 6, 1).toDateTimeAtStartOfDay(DateTimeZone.UTC), new LocalDate(2012, 7, 1).toDateTimeAtStartOfDay(DateTimeZone.UTC), new BigDecimal("350"));
+        usageList.add(usage3);
+        setUsage(usageList);
+
+        busHandler.pushExpectedEvents(NextEvent.INVOICE, NextEvent.PAYMENT);
+        clock.setDay(new LocalDate(2012, 7, 1));
+        assertListenerStatus();
+
+        invoiceChecker.checkInvoice(account.getId(), 4, callContext,
+                                    new ExpectedInvoiceItemCheck(new LocalDate(2012, 6, 1), new LocalDate(2012, 7, 1), InvoiceItemType.USAGE, new BigDecimal("11.80")));
+
     }
 
-    private void setUsage(final RolledUpUsage...usages) {
-        final List<RolledUpUsage> usageList = new ArrayList<RolledUpUsage>();
-        for (RolledUpUsage usage : usages) {
-            usageList.add(usage);
-        }
+    private void setUsage(final List<RolledUpUsage> usageList) {
         ((MockUsageUserApi) usageUserApi).setAllUsageForSubscription(usageList);
     }
 }

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/generator/DefaultInvoiceGenerator.java b/invoice/src/main/java/org/killbill/billing/invoice/generator/DefaultInvoiceGenerator.java
index 5e2ebe0..6b4d6d8 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/generator/DefaultInvoiceGenerator.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/generator/DefaultInvoiceGenerator.java
@@ -105,7 +105,7 @@ public class DefaultInvoiceGenerator implements InvoiceGenerator {
         final List<InvoiceItem> usageItems = generateUsageInvoiceItems(invoiceId, events, existingInvoices, targetDate, context);
         invoice.addInvoiceItems(usageItems);
 
-        return inAdvanceItems.size() != 0 ? invoice : null;
+        return invoice.getInvoiceItems().size() != 0 ? invoice : null;
     }
 
     // STEPH_USAGE Only deals with consumable in arrear usage billing.
@@ -123,10 +123,16 @@ public class DefaultInvoiceGenerator implements InvoiceGenerator {
             UUID curSubscriptionId = null;
             while (events.hasNext()) {
                 final BillingEvent event = events.next();
+                // Skip events that are posterior to the targetDate
+                final LocalDate eventLocalEffectiveDate =  new LocalDate(event.getEffectiveDate(), event.getAccount().getTimeZone());
+                if (eventLocalEffectiveDate.isAfter(targetDate)) {
+                    continue;
+                }
+
                 final UUID subscriptionId = event.getSubscription().getId();
                 if (curSubscriptionId != null && !curSubscriptionId.equals(subscriptionId)) {
                     final SubscriptionConsumableInArrear subscriptionConsumableInArrear = new SubscriptionConsumableInArrear(invoiceId, curEvents, usageApi, targetDate, context.toTenantContext(tenantId));
-                    items.addAll(subscriptionConsumableInArrear.computeMissingUsageInvoiceItems(extractUsageItemsForSubscription(subscriptionId, existingInvoices)));
+                    items.addAll(subscriptionConsumableInArrear.computeMissingUsageInvoiceItems(extractUsageItemsForSubscription(curSubscriptionId, existingInvoices)));
                     curEvents = Lists.newArrayList();
                 }
                 curSubscriptionId = subscriptionId;

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/InvoiceDispatcher.java b/invoice/src/main/java/org/killbill/billing/invoice/InvoiceDispatcher.java
index 24e9132..3b74980 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/InvoiceDispatcher.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/InvoiceDispatcher.java
@@ -244,7 +244,7 @@ public class InvoiceDispatcher {
                                                                                                                                                          }
                                                                                                                                                      }));
 
-                    final Map<UUID, List<DateTime>> callbackDateTimePerSubscriptions = createNextFutureNotificationDate(invoiceItemModelDaos, UsageUtils.getKnownUsages(billingEvents, null), dateAndTimeZoneContext);
+                    final Map<UUID, List<DateTime>> callbackDateTimePerSubscriptions = createNextFutureNotificationDate(invoiceItemModelDaos, billingEvents.getUsages(), dateAndTimeZoneContext);
                     invoiceDao.createInvoice(invoiceModelDao, invoiceItemModelDaos, invoicePaymentModelDaos, isRealInvoiceWithItems, callbackDateTimePerSubscriptions, context);
 
                     final List<InvoiceItem> fixedPriceInvoiceItems = invoice.getInvoiceItems(FixedPriceInvoiceItem.class);
@@ -317,8 +317,7 @@ public class InvoiceDispatcher {
                     break;
 
                 case USAGE:
-                    final Usage usage = knownUsages.get(item.getUsageName());
-                    final String key = item.getSubscriptionId().toString() + ":" + usage.getName();
+                    final String key = item.getSubscriptionId().toString() + ":" + item.getUsageName();
                     final LocalDate perSubscriptionUsageRecurringDate  = perSubscriptionUsage.get(key);
                     if (perSubscriptionUsageRecurringDate == null || perSubscriptionUsageRecurringDate.compareTo(item.getEndDate()) < 0) {
                         perSubscriptionUsage.put(key, item.getEndDate());
@@ -336,17 +335,21 @@ public class InvoiceDispatcher {
 
             final List<DateTime> perSubscriptionCallback = result.get(subscriptionId);
             final String usageName = parts[1];
-            final Usage usage = knownUsages.get(usageName);
-
             final LocalDate endDate =  perSubscriptionUsage.get(key);
-            // STEPH_USAGE WE should double check this is indeed the right date to be called back for that subscription/usage section.
-            final LocalDate nextCallbackUsageDate = (usage.getBillingMode() == BillingMode.IN_ARREAR) ? endDate.plusMonths(usage.getBillingPeriod().getNumberOfMonths()) : endDate;
-            perSubscriptionCallback.add(dateAndTimeZoneContext.computeUTCDateTimeFromLocalDate(nextCallbackUsageDate));
+
+            final DateTime subscriptionUsageCallbackDate = getNextUsageBillingDate(usageName, endDate, dateAndTimeZoneContext, knownUsages);
+            perSubscriptionCallback.add(subscriptionUsageCallbackDate);
         }
 
         return result;
     }
 
+    private DateTime getNextUsageBillingDate(final String usageName, final LocalDate chargedThroughDate, final DateAndTimeZoneContext dateAndTimeZoneContext, final Map<String, Usage> knownUsages) {
+        final Usage usage = knownUsages.get(usageName);
+        final LocalDate nextCallbackUsageDate = (usage.getBillingMode() == BillingMode.IN_ARREAR) ? chargedThroughDate.plusMonths(usage.getBillingPeriod().getNumberOfMonths()) : chargedThroughDate;
+        return dateAndTimeZoneContext.computeUTCDateTimeFromLocalDate(nextCallbackUsageDate);
+    }
+
     private void setChargedThroughDates(final DateAndTimeZoneContext dateAndTimeZoneContext,
                                         final Collection<InvoiceItem> fixedPriceItems,
                                         final Collection<InvoiceItem> recurringItems,

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/usage/ContiguousIntervalConsumableInArrear.java b/invoice/src/main/java/org/killbill/billing/invoice/usage/ContiguousIntervalConsumableInArrear.java
index 4e016f7..2ee9459 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/usage/ContiguousIntervalConsumableInArrear.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/usage/ContiguousIntervalConsumableInArrear.java
@@ -102,7 +102,6 @@ public class ContiguousIntervalConsumableInArrear {
         if (targetDate.isBefore(startDate)) {
             return this;
         }
-        // STEPH_USAGE should we billed up to endDate to targetDate is targetDate < endDate./ What does targetDate means in arrear
         final LocalDate endDate = closedInterval ? new LocalDate(billingEvents.get(billingEvents.size() - 1).getEffectiveDate(), getAccountTimeZone()) : targetDate;
 
         final BillingIntervalDetail bid = new BillingIntervalDetail(startDate, endDate, targetDate, getBCD(), usage.getBillingPeriod());
@@ -147,10 +146,12 @@ public class ContiguousIntervalConsumableInArrear {
                 toBeBilledUsage = toBeBilledUsage.add(toBeBilledForUnit);
             }
             // Retrieves current price amount billed for that period of time (and usage section)
-            final BigDecimal billedUsage = computeBilledUsage(ru.getStartDate(), ru.getEndDate(), existingUsage);
+            final Iterable<InvoiceItem> billedItems = getBilledItems(ru.getStartDate(), ru.getEndDate(), existingUsage);
+            final BigDecimal billedUsage = computeBilledUsage(billedItems);
 
-            // Compare the two and add the missing piece if required.
-            if (billedUsage.compareTo(toBeBilledUsage) < 0) {
+            // Compare the two and add the missing piece if required. If there has never been any billed item for the period
+            // and if there is nothing to bill for we would also insert a $0 amount
+            if (!billedItems.iterator().hasNext() || billedUsage.compareTo(toBeBilledUsage) < 0) {
                 InvoiceItem item = new UsageInvoiceItem(invoiceId, getAccountId(), getBundleId(), getSubscriptionId(), getPlanName(),
                                                         getPhaseName(), usage.getName(), ru.getStartDate(), ru.getEndDate(), toBeBilledUsage.subtract(billedUsage), getCurrency());
                 result.add(item);
@@ -165,6 +166,10 @@ public class ContiguousIntervalConsumableInArrear {
      * @return a list of {@code RolledUpUsage} for each period (between two transitions) * each unitType.
      */
     List<RolledUpUsage> getRolledUpUsage() {
+        // There needs to be at least two transitions to define an interval to bill
+        if (transitionTimes.size() <= 1) {
+            return Collections.emptyList();
+        }
 
         final Iterable<DateTime> transitions = Iterables.transform(transitionTimes, new Function<LocalDate, DateTime>() {
             @Override
@@ -172,7 +177,6 @@ public class ContiguousIntervalConsumableInArrear {
                 return localDateToEndOfDayInAccountTimezone(input, getAccountTimeZone());
             }
         });
-        // STEPH_USAGE optimized api takes set of unitTypes -- for usage section-- and list of transitions date. Should we use dateTime or LocalDate?
         return usageApi.getAllUsageForSubscription(getSubscriptionId(), unitTypes, ImmutableList.copyOf(transitions), context);
     }
 
@@ -210,36 +214,38 @@ public class ContiguousIntervalConsumableInArrear {
 
     /**
      *
-     * @param startDate the startDate of of billed period
-     * @param endDate the endDate of of billed period
-     * @param existingUsage the list of invoiceItem for that subscription
+     * @param filteredUsageForInterval the list of invoiceItem to consider
      * @return the price amount that was already billed for that period and usage section (across unitTypes)
      */
     @VisibleForTesting
-    BigDecimal computeBilledUsage(final LocalDate startDate, final LocalDate endDate, final List<InvoiceItem> existingUsage) {
+    BigDecimal computeBilledUsage(final Iterable<InvoiceItem> filteredUsageForInterval) {
+
+        Preconditions.checkState(isBuilt.get());
+        BigDecimal billedAmount = BigDecimal.ZERO;
+        for (InvoiceItem ii : filteredUsageForInterval) {
+            billedAmount = billedAmount.add(ii.getAmount());
+        }
+        // Return the billed $ amount (not the # of units)
+        return billedAmount;
+    }
+
+    Iterable<InvoiceItem> getBilledItems(final LocalDate startDate, final LocalDate endDate, final List<InvoiceItem> existingUsage) {
 
         Preconditions.checkState(isBuilt.get());
-        final Iterable<InvoiceItem> filteredUsageForInterval = Iterables.filter(existingUsage, new Predicate<InvoiceItem>() {
+        return Iterables.filter(existingUsage, new Predicate<InvoiceItem>() {
             @Override
             public boolean apply(final InvoiceItem input) {
                 if (input.getInvoiceItemType() != InvoiceItemType.USAGE) {
                     return false;
                 }
 
-                // STEPH_USAGE what happens if we discover usage period that overlap (one side or both side) the [startDate, endDate] interval
+                // STEPH what happens if we discover usage period that overlap (one side or both side) the [startDate, endDate] interval
                 final UsageInvoiceItem usageInput = (UsageInvoiceItem) input;
                 return usageInput.getUsageName().equals(usage.getName()) &&
                        usageInput.getStartDate().compareTo(startDate) >= 0 &&
                        usageInput.getEndDate().compareTo(endDate) <= 0;
             }
         });
-
-        BigDecimal billedAmount = BigDecimal.ZERO;
-        for (InvoiceItem ii : filteredUsageForInterval) {
-            billedAmount = billedAmount.add(ii.getAmount());
-        }
-        // Return the billed $ amount (not the # of units)
-        return billedAmount;
     }
 
     @VisibleForTesting

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/usage/UsageUtils.java b/invoice/src/main/java/org/killbill/billing/invoice/usage/UsageUtils.java
index a96b80e..b3a1f5a 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/usage/UsageUtils.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/usage/UsageUtils.java
@@ -42,25 +42,6 @@ import com.google.common.collect.Lists;
 
 public class UsageUtils {
 
-    public static Map<String, Usage> getKnownUsages(final BillingEventSet billingEvents, @Nullable final Predicate filter) {
-        final Iterable<Usage> usages = Iterables.concat(Iterables.transform(billingEvents, new Function<BillingEvent, List<Usage>>() {
-            @Override
-            public List<Usage> apply(final BillingEvent input) {
-                return input.getUsages();
-            }
-        }));
-
-        final Iterable<Usage> filteredUsages = (filter != null) ? Iterables.filter(usages, filter) : usages;
-
-        final Map<String, Usage> result = (filteredUsages.iterator().hasNext()) ? new HashMap<String, Usage>() : Collections.<String, Usage>emptyMap();
-        final Iterator<Usage> iterator = filteredUsages.iterator();
-        while (iterator.hasNext()) {
-            final Usage next = iterator.next();
-            result.put(next.getName(), next);
-        }
-        return result;
-    }
-
     public static List<TieredBlock> getConsumableInArrearTieredBlocks(final Usage usage, final String unitType) {
 
         Preconditions.checkArgument(usage.getBillingMode() == BillingMode.IN_ARREAR && usage.getUsageType() == UsageType.CONSUMABLE);

diff --git a/invoice/src/test/java/org/killbill/billing/invoice/MockBillingEventSet.java b/invoice/src/test/java/org/killbill/billing/invoice/MockBillingEventSet.java
index 261122f..d93cf5b 100644
--- a/invoice/src/test/java/org/killbill/billing/invoice/MockBillingEventSet.java
+++ b/invoice/src/test/java/org/killbill/billing/invoice/MockBillingEventSet.java
@@ -17,11 +17,14 @@
 package org.killbill.billing.invoice;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.TreeSet;
 import java.util.UUID;
 
 import org.killbill.billing.catalog.api.BillingMode;
+import org.killbill.billing.catalog.api.Usage;
 import org.killbill.billing.junction.BillingEvent;
 import org.killbill.billing.junction.BillingEventSet;
 
@@ -51,6 +54,11 @@ public class MockBillingEventSet extends TreeSet<BillingEvent> implements Billin
         return subscriptionIdsWithAutoInvoiceOff;
     }
 
+    @Override
+    public Map<String, Usage> getUsages() {
+        return Collections.emptyMap();
+    }
+
     public void setAccountInvoiceOff(final boolean isAccountInvoiceOff) {
         this.isAccountInvoiceOff = isAccountInvoiceOff;
     }

diff --git a/invoice/src/test/java/org/killbill/billing/invoice/TestInvoiceHelper.java b/invoice/src/test/java/org/killbill/billing/invoice/TestInvoiceHelper.java
index 91a4385..22e1331 100644
--- a/invoice/src/test/java/org/killbill/billing/invoice/TestInvoiceHelper.java
+++ b/invoice/src/test/java/org/killbill/billing/invoice/TestInvoiceHelper.java
@@ -313,10 +313,14 @@ public class TestInvoiceHelper {
                                                final BillingMode billingMode, final String description,
                                                final long totalOrdering,
                                                final SubscriptionBaseTransitionType type) {
+
+        final Account mockAccount = Mockito.mock(Account.class);
+        Mockito.when(mockAccount.getTimeZone()).thenReturn(DateTimeZone.UTC);
+        final Account accountOrMockAcount = account != null ? account : mockAccount;
         return new BillingEvent() {
             @Override
             public Account getAccount() {
-                return account;
+                return accountOrMockAcount;
             }
 
             @Override

diff --git a/invoice/src/test/java/org/killbill/billing/invoice/usage/TestContiguousIntervalConsumableInArrear.java b/invoice/src/test/java/org/killbill/billing/invoice/usage/TestContiguousIntervalConsumableInArrear.java
index 0e9d662..a78dcbc 100644
--- a/invoice/src/test/java/org/killbill/billing/invoice/usage/TestContiguousIntervalConsumableInArrear.java
+++ b/invoice/src/test/java/org/killbill/billing/invoice/usage/TestContiguousIntervalConsumableInArrear.java
@@ -92,7 +92,7 @@ public class TestContiguousIntervalConsumableInArrear extends TestUsageInArrearB
         final FixedPriceInvoiceItem ii5 = new FixedPriceInvoiceItem(invoiceId, accountId, bundleId, subscriptionId, planName, phaseName, startDate, BigDecimal.TEN, currency);
         existingUsage.add(ii5);
 
-        final BigDecimal result = intervalConsumableInArrear.computeBilledUsage(startDate, endDate, existingUsage);
+        final BigDecimal result = intervalConsumableInArrear.computeBilledUsage(intervalConsumableInArrear.getBilledItems(startDate, endDate, existingUsage));
         assertEquals(result, BigDecimal.TEN.add(BigDecimal.TEN));
     }
 

diff --git a/junction/src/main/java/org/killbill/billing/junction/plumbing/billing/DefaultBillingEventSet.java b/junction/src/main/java/org/killbill/billing/junction/plumbing/billing/DefaultBillingEventSet.java
index 87b8231..4586db9 100644
--- a/junction/src/main/java/org/killbill/billing/junction/plumbing/billing/DefaultBillingEventSet.java
+++ b/junction/src/main/java/org/killbill/billing/junction/plumbing/billing/DefaultBillingEventSet.java
@@ -17,15 +17,28 @@
 package org.killbill.billing.junction.plumbing.billing;
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.UUID;
 
+import javax.annotation.Nullable;
+
 import org.killbill.billing.catalog.api.BillingMode;
+import org.killbill.billing.catalog.api.Usage;
 import org.killbill.billing.junction.BillingEvent;
 import org.killbill.billing.junction.BillingEventSet;
 
+import com.google.common.base.Function;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
 public class DefaultBillingEventSet extends TreeSet<BillingEvent> implements SortedSet<BillingEvent>, BillingEventSet {
 
     private static final long serialVersionUID = 1L;
@@ -55,6 +68,24 @@ public class DefaultBillingEventSet extends TreeSet<BillingEvent> implements Sor
         return subscriptionIdsWithAutoInvoiceOff;
     }
 
+    @Override
+    public Map<String, Usage> getUsages() {
+        final Iterable<Usage> allUsages = Iterables.concat(Iterables.transform(this, new Function<BillingEvent, List<Usage>>() {
+            @Override
+            public List<Usage> apply(final BillingEvent input) {
+                return input.getUsages();
+            }
+        }));
+        if (!allUsages.iterator().hasNext()) {
+            return Collections.emptyMap();
+        }
+        final Map<String, Usage> result = new HashMap<String, Usage>();
+        for (Usage cur : Sets.<Usage>newHashSet(allUsages)) {
+            result.put(cur.getName(), cur);
+        }
+        return result;
+    }
+
     public void setAccountAutoInvoiceIsOff(final boolean accountAutoInvoiceIsOff) {
         this.accountAutoInvoiceOff = accountAutoInvoiceIsOff;
     }

diff --git a/junction/src/main/java/org/killbill/billing/junction/plumbing/billing/DefaultInternalBillingApi.java b/junction/src/main/java/org/killbill/billing/junction/plumbing/billing/DefaultInternalBillingApi.java
index 1cbcbf6..daa9a47 100644
--- a/junction/src/main/java/org/killbill/billing/junction/plumbing/billing/DefaultInternalBillingApi.java
+++ b/junction/src/main/java/org/killbill/billing/junction/plumbing/billing/DefaultInternalBillingApi.java
@@ -22,6 +22,13 @@ import java.util.UUID;
 
 import javax.annotation.Nullable;
 
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.joda.time.LocalDate;
+import org.killbill.billing.catalog.api.BillingMode;
+import org.killbill.billing.catalog.api.Usage;
+import org.killbill.billing.util.timezone.DateAndTimeZoneContext;
+import org.killbill.clock.Clock;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -57,19 +64,23 @@ public class DefaultInternalBillingApi implements BillingInternalApi {
     private final CatalogService catalogService;
     private final BlockingCalculator blockCalculator;
     private final TagInternalApi tagApi;
+    private final Clock clock;
 
     @Inject
     public DefaultInternalBillingApi(final AccountInternalApi accountApi,
                                      final BillCycleDayCalculator bcdCalculator,
                                      final SubscriptionBaseInternalApi subscriptionApi,
                                      final BlockingCalculator blockCalculator,
-                                     final CatalogService catalogService, final TagInternalApi tagApi) {
+                                     final CatalogService catalogService,
+                                     final TagInternalApi tagApi,
+                                     final Clock clock) {
         this.accountApi = accountApi;
         this.bcdCalculator = bcdCalculator;
         this.subscriptionApi = subscriptionApi;
         this.catalogService = catalogService;
         this.blockCalculator = blockCalculator;
         this.tagApi = tagApi;
+        this.clock = clock;
     }
 
     @Override