killbill-aplcache

diff --git a/api/src/main/java/org/killbill/billing/usage/InternalUserApi.java b/api/src/main/java/org/killbill/billing/usage/InternalUserApi.java
new file mode 100644
index 0000000..089a208
--- /dev/null
+++ b/api/src/main/java/org/killbill/billing/usage/InternalUserApi.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.usage;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.joda.time.LocalDate;
+import org.killbill.billing.callcontext.InternalTenantContext;
+
+public interface InternalUserApi {
+
+    public List<RawUsage> getRawUsageForAccount(final UUID accountId, final LocalDate stateDate, final LocalDate endDate, final InternalTenantContext tenantContext);
+}

diff --git a/api/src/main/java/org/killbill/billing/usage/RawUsage.java b/api/src/main/java/org/killbill/billing/usage/RawUsage.java
new file mode 100644
index 0000000..b57a3dc
--- /dev/null
+++ b/api/src/main/java/org/killbill/billing/usage/RawUsage.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.usage;
+
+import java.util.UUID;
+
+import org.joda.time.LocalDate;
+
+public interface RawUsage {
+
+    UUID getSubscriptionId();
+
+    LocalDate getDate();
+
+    String getUnitType();
+
+    Long getAmount();
+}

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 b5096a2..d7896b0 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
@@ -19,11 +19,8 @@ package org.killbill.billing.beatrix.integration.usage;
 import java.math.BigDecimal;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Set;
 import java.util.UUID;
 
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
 import org.joda.time.LocalDate;
 import org.killbill.billing.account.api.Account;
 import org.killbill.billing.account.api.AccountData;
@@ -34,14 +31,10 @@ import org.killbill.billing.catalog.api.BillingPeriod;
 import org.killbill.billing.catalog.api.ProductCategory;
 import org.killbill.billing.entitlement.api.DefaultEntitlement;
 import org.killbill.billing.invoice.api.InvoiceItemType;
-import org.killbill.billing.usage.api.RolledUpUsage;
 import org.killbill.billing.usage.api.SubscriptionUsageRecord;
 import org.killbill.billing.usage.api.UnitUsageRecord;
 import org.killbill.billing.usage.api.UsageRecord;
-import org.killbill.billing.usage.api.UsageUserApi;
 import org.killbill.billing.util.callcontext.CallContext;
-import org.killbill.billing.util.callcontext.TenantContext;
-import org.mockito.Mockito;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
@@ -52,16 +45,9 @@ public class TestConsumableInArrear extends TestIntegrationBase {
         super.beforeMethod();
     }
 
-    protected UsageUserApi createMockUsageUserApi(final List<RolledUpUsage> returnValue) {
-        final UsageUserApi result = Mockito.mock(UsageUserApi.class);
-        Mockito.when(result.getAllUsageForSubscription(Mockito.<UUID>any(), Mockito.<List<LocalDate>>any(), Mockito.<TenantContext>any())).thenReturn(returnValue);
-        return result;
-    }
-
     @Test(groups = "slow")
     public void testSimple() throws Exception {
 
-
         final AccountData accountData = getAccountData(1);
         final Account account = createAccountWithNonOsgiPaymentMethod(accountData);
         accountChecker.checkAccount(account.getId(), accountData, callContext);
@@ -78,7 +64,6 @@ public class TestConsumableInArrear extends TestIntegrationBase {
         subscriptionChecker.checkSubscriptionCreated(bpSubscription.getId(), internalCallContext);
         invoiceChecker.checkInvoice(account.getId(), 1, callContext, new ExpectedInvoiceItemCheck(new LocalDate(2012, 4, 1), null, InvoiceItemType.FIXED, new BigDecimal("0")));
 
-
         //
         // ADD ADD_ON ON THE SAME DAY
         //
@@ -95,24 +80,18 @@ public class TestConsumableInArrear extends TestIntegrationBase {
                                     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")));
 
-
-        busHandler.pushExpectedEvents(NextEvent.INVOICE);
+        // We don't expect any invoice, but we want to give the system the time to verify there is nothing to do so we can fail
         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));
-
+        Thread.sleep(1000);
 
         setUsage(aoSubscription.getId(), "bullets", new LocalDate(2012, 6, 1), 50L, callContext);
         setUsage(aoSubscription.getId(), "bullets", new LocalDate(2012, 6, 16), 300L, callContext);
 
-
         busHandler.pushExpectedEvents(NextEvent.INVOICE, NextEvent.PAYMENT);
         clock.setDay(new LocalDate(2012, 7, 1));
         assertListenerStatus();
 
-        invoiceChecker.checkInvoice(account.getId(), 4, callContext,
+        invoiceChecker.checkInvoice(account.getId(), 3, callContext,
                                     new ExpectedInvoiceItemCheck(new LocalDate(2012, 6, 1), new LocalDate(2012, 7, 1), InvoiceItemType.USAGE, new BigDecimal("11.80")));
 
     }

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/dao/DefaultInvoiceDao.java b/invoice/src/main/java/org/killbill/billing/invoice/dao/DefaultInvoiceDao.java
index 022da79..9359392 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/dao/DefaultInvoiceDao.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/dao/DefaultInvoiceDao.java
@@ -220,6 +220,10 @@ public class DefaultInvoiceDao extends EntityDaoBase<InvoiceModelDao, Invoice, I
     public void createInvoice(final InvoiceModelDao invoice, final List<InvoiceItemModelDao> invoiceItems,
                               final boolean isRealInvoice, final FutureAccountNotifications callbackDateTimePerSubscriptions,
                               final InternalCallContext context) {
+
+        // We could be called with an empty list of items (for when we ONLY need to set the future account notifications).
+        final boolean hasInvoiceItems = !invoiceItems.isEmpty();
+
         transactionalSqlDao.execute(new EntitySqlDaoTransactionWrapper<Void>() {
             @Override
             public Void inTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory) throws Exception {
@@ -233,14 +237,15 @@ public class DefaultInvoiceDao extends EntityDaoBase<InvoiceModelDao, Invoice, I
                         transactional.create(invoice, context);
                     }
 
-                    // Create the invoice items
-                    final InvoiceItemSqlDao transInvoiceItemSqlDao = entitySqlDaoWrapperFactory.become(InvoiceItemSqlDao.class);
-                    for (final InvoiceItemModelDao invoiceItemModelDao : invoiceItems) {
-                        createInvoiceItemFromTransaction(transInvoiceItemSqlDao, invoiceItemModelDao, context);
+                    if (hasInvoiceItems) {
+                        // Create the invoice items
+                        final InvoiceItemSqlDao transInvoiceItemSqlDao = entitySqlDaoWrapperFactory.become(InvoiceItemSqlDao.class);
+                        for (final InvoiceItemModelDao invoiceItemModelDao : invoiceItems) {
+                            createInvoiceItemFromTransaction(transInvoiceItemSqlDao, invoiceItemModelDao, context);
+                        }
+                        cbaDao.addCBAComplexityFromTransaction(invoice, entitySqlDaoWrapperFactory, context);
                     }
 
-                    cbaDao.addCBAComplexityFromTransaction(invoice, entitySqlDaoWrapperFactory, context);
-
                     notifyOfFutureBillingEvents(entitySqlDaoWrapperFactory, invoice.getAccountId(), callbackDateTimePerSubscriptions, context);
                 }
                 return null;

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 363f012..c298495 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
@@ -36,6 +36,8 @@ import org.killbill.billing.catalog.api.BillingMode;
 import org.killbill.billing.catalog.api.BillingPeriod;
 import org.killbill.billing.catalog.api.CatalogApiException;
 import org.killbill.billing.catalog.api.Currency;
+import org.killbill.billing.catalog.api.Usage;
+import org.killbill.billing.catalog.api.UsageType;
 import org.killbill.billing.invoice.api.Invoice;
 import org.killbill.billing.invoice.api.InvoiceApiException;
 import org.killbill.billing.invoice.api.InvoiceItem;
@@ -51,7 +53,8 @@ import org.killbill.billing.invoice.tree.AccountItemTree;
 import org.killbill.billing.invoice.usage.SubscriptionConsumableInArrear;
 import org.killbill.billing.junction.BillingEvent;
 import org.killbill.billing.junction.BillingEventSet;
-import org.killbill.billing.usage.api.UsageUserApi;
+import org.killbill.billing.usage.InternalUserApi;
+import org.killbill.billing.usage.RawUsage;
 import org.killbill.billing.util.callcontext.InternalCallContextFactory;
 import org.killbill.billing.util.callcontext.TenantContext;
 import org.killbill.billing.util.config.InvoiceConfig;
@@ -73,15 +76,13 @@ public class DefaultInvoiceGenerator implements InvoiceGenerator {
 
     private final Clock clock;
     private final InvoiceConfig config;
-    private final UsageUserApi usageApi;
-    private final InternalCallContextFactory internalCallContextFactory;
+    private final InternalUserApi usageApi;
 
     @Inject
-    public DefaultInvoiceGenerator(final Clock clock, final UsageUserApi usageApi, final InvoiceConfig config, final InternalCallContextFactory internalCallContextFactory) {
+    public DefaultInvoiceGenerator(final Clock clock, final InternalUserApi usageApi, final InvoiceConfig config) {
         this.clock = clock;
         this.config = config;
         this.usageApi = usageApi;
-        this.internalCallContextFactory = internalCallContextFactory;
     }
 
     /*
@@ -105,22 +106,23 @@ public class DefaultInvoiceGenerator implements InvoiceGenerator {
         final List<InvoiceItem> inAdvanceItems = generateInAdvanceInvoiceItems(account.getId(), invoiceId, events, existingInvoices, adjustedTargetDate, targetCurrency);
         invoice.addInvoiceItems(inAdvanceItems);
 
-        final List<InvoiceItem> usageItems = generateUsageInvoiceItems(invoiceId, events, existingInvoices, targetDate, context);
+        final List<InvoiceItem> usageItems = generateUsageInvoiceItems(account, invoiceId, events, existingInvoices, targetDate, context);
         invoice.addInvoiceItems(usageItems);
 
         return invoice.getInvoiceItems().size() != 0 ? invoice : null;
     }
 
     // STEPH_USAGE Only deals with consumable in arrear usage billing.
-    private List<InvoiceItem> generateUsageInvoiceItems(final UUID invoiceId, final BillingEventSet eventSet,
+    private List<InvoiceItem> generateUsageInvoiceItems(final Account account,
+                                                        final UUID invoiceId, final BillingEventSet eventSet,
                                                         @Nullable final List<Invoice> existingInvoices, final LocalDate targetDate,
-                                                        final InternalCallContext context) throws InvoiceApiException {
-        final TenantContext tenantContext = internalCallContextFactory.createTenantContext(context);
+                                                        final InternalCallContext internalCallContext) throws InvoiceApiException {
         try {
-
             final List<InvoiceItem> items = Lists.newArrayList();
             final Iterator<BillingEvent> events = eventSet.iterator();
 
+            boolean seenAnyUsageItems = false;
+            List<RawUsage> rawUsage = ImmutableList.of();
             List<BillingEvent> curEvents = Lists.newArrayList();
             UUID curSubscriptionId = null;
             while (events.hasNext()) {
@@ -131,9 +133,25 @@ public class DefaultInvoiceGenerator implements InvoiceGenerator {
                     continue;
                 }
 
+                // Optimize to do the usage query only once after we know there are indeed some usage items
+                if (!seenAnyUsageItems) {
+                    final boolean foundUsage = Iterables.tryFind(event.getUsages(), new Predicate<Usage>() {
+                        @Override
+                        public boolean apply(@Nullable final Usage input) {
+                            return (input.getUsageType() == UsageType.CONSUMABLE &&
+                                    input.getBillingMode() == BillingMode.IN_ARREAR);
+                        }
+                    }).orNull() != null;
+                    if (foundUsage) {
+                        rawUsage = usageApi.getRawUsageForAccount(account.getId(), new LocalDate(event.getEffectiveDate(), account.getTimeZone()), targetDate, internalCallContext);
+                        seenAnyUsageItems = true;
+                    }
+                }
+
+                // We always go in the usage invoicing logic to at least generate $0 USAGE items that will indicate the time for the next notification
                 final UUID subscriptionId = event.getSubscription().getId();
                 if (curSubscriptionId != null && !curSubscriptionId.equals(subscriptionId)) {
-                    final SubscriptionConsumableInArrear subscriptionConsumableInArrear = new SubscriptionConsumableInArrear(invoiceId, curEvents, usageApi, config.isInsertZeroUsageItems(), targetDate, tenantContext);
+                    final SubscriptionConsumableInArrear subscriptionConsumableInArrear = new SubscriptionConsumableInArrear(invoiceId, curEvents, rawUsage, targetDate);
                     items.addAll(subscriptionConsumableInArrear.computeMissingUsageInvoiceItems(extractUsageItemsForSubscription(curSubscriptionId, existingInvoices)));
                     curEvents = Lists.newArrayList();
                 }
@@ -141,7 +159,8 @@ public class DefaultInvoiceGenerator implements InvoiceGenerator {
                 curEvents.add(event);
             }
             if (curSubscriptionId != null) {
-                final SubscriptionConsumableInArrear subscriptionConsumableInArrear = new SubscriptionConsumableInArrear(invoiceId, curEvents, usageApi, config.isInsertZeroUsageItems(), targetDate, tenantContext);
+
+                final SubscriptionConsumableInArrear subscriptionConsumableInArrear = new SubscriptionConsumableInArrear(invoiceId, curEvents, rawUsage, targetDate);
                 items.addAll(subscriptionConsumableInArrear.computeMissingUsageInvoiceItems(extractUsageItemsForSubscription(curSubscriptionId, existingInvoices)));
             }
             return items;

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 1686c1f..9b9274d 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/InvoiceDispatcher.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/InvoiceDispatcher.java
@@ -90,8 +90,10 @@ import org.slf4j.LoggerFactory;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Function;
 import com.google.common.base.Joiner;
+import com.google.common.base.Predicate;
 import com.google.common.collect.Collections2;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
 import com.google.inject.Inject;
 
 public class InvoiceDispatcher {
@@ -279,16 +281,26 @@ public class InvoiceDispatcher {
 
                 // Transformation to Invoice -> InvoiceModelDao
                 final InvoiceModelDao invoiceModelDao = new InvoiceModelDao(invoice);
-                final List<InvoiceItemModelDao> invoiceItemModelDaos = ImmutableList.copyOf(Collections2.transform(invoice.getInvoiceItems(),
-                                                                                                                   new Function<InvoiceItem, InvoiceItemModelDao>() {
-                                                                                                                       @Override
-                                                                                                                       public InvoiceItemModelDao apply(final InvoiceItem input) {
-                                                                                                                           return new InvoiceItemModelDao(input);
-                                                                                                                       }
-                                                                                                                   }));
-
+                final Iterable<InvoiceItemModelDao> invoiceItemModelDaos = Iterables.transform(invoice.getInvoiceItems(),
+                                                                                               new Function<InvoiceItem, InvoiceItemModelDao>() {
+                                                                                                   @Override
+                                                                                                   public InvoiceItemModelDao apply(final InvoiceItem input) {
+                                                                                                       return new InvoiceItemModelDao(input);
+                                                                                                   }
+                                                                                               });
                 final FutureAccountNotifications futureAccountNotifications = createNextFutureNotificationDate(invoiceItemModelDaos, billingEvents.getUsages(), dateAndTimeZoneContext);
-                invoiceDao.createInvoice(invoiceModelDao, invoiceItemModelDaos, isRealInvoiceWithItems, futureAccountNotifications, context);
+
+                // We filter any zero amount for USAGE items prior we generate the invoice, which may leave us with an invoice with no items;
+                // we recompute the isRealInvoiceWithItems flag based on what is left (the call to invoice is still necessary to set the future notifications).
+                final Iterable<InvoiceItemModelDao> filteredInvoiceItemModelDaos = Iterables.filter(invoiceItemModelDaos, new Predicate<InvoiceItemModelDao>() {
+                    @Override
+                    public boolean apply(@Nullable final InvoiceItemModelDao input) {
+                        return (input.getType() != InvoiceItemType.USAGE || input.getAmount().compareTo(BigDecimal.ZERO) != 0);
+                    }
+                });
+                isRealInvoiceWithItems = filteredInvoiceItemModelDaos.iterator().hasNext() ? isRealInvoiceWithItems : false;
+
+                invoiceDao.createInvoice(invoiceModelDao, ImmutableList.copyOf(filteredInvoiceItemModelDaos), isRealInvoiceWithItems, futureAccountNotifications, context);
 
                 final List<InvoiceItem> fixedPriceInvoiceItems = invoice.getInvoiceItems(FixedPriceInvoiceItem.class);
                 final List<InvoiceItem> recurringInvoiceItems = invoice.getInvoiceItems(RecurringInvoiceItem.class);
@@ -355,7 +367,7 @@ public class InvoiceDispatcher {
 
 
     @VisibleForTesting
-    FutureAccountNotifications createNextFutureNotificationDate(final List<InvoiceItemModelDao> invoiceItems, final Map<String, Usage> knownUsages, final DateAndTimeZoneContext dateAndTimeZoneContext) {
+    FutureAccountNotifications createNextFutureNotificationDate(final Iterable<InvoiceItemModelDao> invoiceItems, final Map<String, Usage> knownUsages, final DateAndTimeZoneContext dateAndTimeZoneContext) {
 
         final Map<UUID, List<DateTime>> result = new HashMap<UUID, List<DateTime>>();
 

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 bbb6de7..2c051fb 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
@@ -17,9 +17,11 @@
 package org.killbill.billing.invoice.usage;
 
 import java.math.BigDecimal;
-import java.util.Collections;
-import java.util.Comparator;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -35,16 +37,16 @@ import org.killbill.billing.invoice.api.InvoiceItemType;
 import org.killbill.billing.invoice.generator.BillingIntervalDetail;
 import org.killbill.billing.invoice.model.UsageInvoiceItem;
 import org.killbill.billing.junction.BillingEvent;
+import org.killbill.billing.usage.RawUsage;
 import org.killbill.billing.usage.api.RolledUpUnit;
 import org.killbill.billing.usage.api.RolledUpUsage;
-import org.killbill.billing.usage.api.UsageUserApi;
-import org.killbill.billing.util.callcontext.TenantContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 
@@ -64,21 +66,17 @@ public class ContiguousIntervalConsumableInArrear {
 
     private final Usage usage;
     private final Set<String> unitTypes;
-    private final UsageUserApi usageApi;
+    private final List<RawUsage> rawSubscriptionUsage;
     private final LocalDate targetDate;
     private final UUID invoiceId;
-    private final TenantContext context;
     private final AtomicBoolean isBuilt;
-    private final boolean insertZeroAmountItems;
 
-    public ContiguousIntervalConsumableInArrear(final Usage usage, final UUID invoiceId, final UsageUserApi usageApi, final boolean insertZeroAmountItems, final LocalDate targetDate, final TenantContext context) {
+    public ContiguousIntervalConsumableInArrear(final Usage usage, final UUID invoiceId, final List<RawUsage> rawSubscriptionUsage, final LocalDate targetDate) {
         this.usage = usage;
         this.invoiceId = invoiceId;
         this.unitTypes = getConsumableInArrearUnitTypes(usage);
-        this.usageApi = usageApi;
-        this.insertZeroAmountItems = insertZeroAmountItems;
+        this.rawSubscriptionUsage = rawSubscriptionUsage;
         this.targetDate = targetDate;
-        this.context = context;
         this.billingEvents = Lists.newLinkedList();
         this.transitionTimes = Lists.newLinkedList();
         this.isBuilt = new AtomicBoolean(false);
@@ -137,8 +135,20 @@ public class ContiguousIntervalConsumableInArrear {
         final List<InvoiceItem> result = Lists.newLinkedList();
 
         final List<RolledUpUsage> allUsage = getRolledUpUsage();
-        for (final RolledUpUsage ru : allUsage) {
 
+        // We start by generating 'marker' USAGE items with $0 that will allow to correctly insert the next notification for when there is no USAGE to bill.
+        // Those will be removed by the invoicing code later so as to not end up with superfluous $0 items
+        LocalDate prevDate = null;
+        for (LocalDate curDate : transitionTimes) {
+            if (prevDate != null) {
+                InvoiceItem item = new UsageInvoiceItem(invoiceId, getAccountId(), getBundleId(), getSubscriptionId(), getPlanName(),
+                                                        getPhaseName(), usage.getName(), prevDate, curDate, BigDecimal.ZERO, getCurrency());
+                result.add(item);
+            }
+            prevDate = curDate;
+        }
+
+        for (final RolledUpUsage ru : allUsage) {
             // Compute total price amount that should be billed for that period of time (and usage section) across unitTypes.
             BigDecimal toBeBilledUsage = BigDecimal.ZERO;
             for (final RolledUpUnit cur : ru.getRolledUpUnits()) {
@@ -158,7 +168,7 @@ public class ContiguousIntervalConsumableInArrear {
             // Compare the two and add the missing piece if required.
             if (!billedItems.iterator().hasNext() || billedUsage.compareTo(toBeBilledUsage) < 0) {
                 final BigDecimal amountToBill = toBeBilledUsage.subtract(billedUsage);
-                if (amountToBill.compareTo(BigDecimal.ZERO) > 0 || insertZeroAmountItems) {
+                if (amountToBill.compareTo(BigDecimal.ZERO) > 0) {
                     InvoiceItem item = new UsageInvoiceItem(invoiceId, getAccountId(), getBundleId(), getSubscriptionId(), getPlanName(),
                                                             getPhaseName(), usage.getName(), ru.getStart(), ru.getEnd(), amountToBill, getCurrency());
                     result.add(item);
@@ -168,15 +178,89 @@ public class ContiguousIntervalConsumableInArrear {
         return result;
     }
 
-    /**
-     * @return a list of {@code RolledUpUsage} for each period (between two transitions) * each unitType.
-     */
+    @VisibleForTesting
     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 Iterator<RawUsage> rawUsageIterator = rawSubscriptionUsage.iterator();
+        if (!rawUsageIterator.hasNext()) {
+            return ImmutableList.of();
+        }
+
+        final List<RolledUpUsage> result = new ArrayList<RolledUpUsage>();
+
+        //
+        // Skip all items before our first transition date
+        //
+        // prevRawUsage keeps track of first unconsumed raw usage element
+        RawUsage prevRawUsage = null;
+        while (rawUsageIterator.hasNext()) {
+            final RawUsage curRawUsage = rawUsageIterator.next();
+            if (curRawUsage.getDate().compareTo(transitionTimes.get(0)) >= 0) {
+                prevRawUsage = curRawUsage;
+                break;
+            }
         }
-        return usageApi.getAllUsageForSubscription(getSubscriptionId(), transitionTimes, context);
+
+        // Optimize path where all raw usage items are outside or our transitionTimes range
+        if (prevRawUsage.getDate().compareTo(transitionTimes.get(transitionTimes.size() - 1)) >= 0) {
+            return ImmutableList.of();
+        }
+
+        //
+        // Loop through each interval [prevDate, curDate) and consume as many rawSubscriptionUsage elements within that range
+        // to create one RolledUpUsage per interval. If an interval does not have any rawSubscriptionUsage element, there will be no
+        // matching RolledUpUsage for that interval, and we'll detect that in the 'computeMissingItems' logic
+        //
+        LocalDate prevDate = null;
+        for (LocalDate curDate : transitionTimes) {
+
+            if (prevDate != null) {
+
+                // Allocate new perRangeUnitToAmount for this interval and populate with rawSubscriptionUsage items
+                final Map<String, Long> perRangeUnitToAmount = new HashMap<String, Long>();
+
+                // Start consuming prevRawUsage element if it exists and falls into the range
+                if (prevRawUsage != null) {
+                    if (prevRawUsage.getDate().compareTo(prevDate) >= 0 && prevRawUsage.getDate().compareTo(curDate) < 0) {
+                        Long currentAmount = perRangeUnitToAmount.get(prevRawUsage.getUnitType());
+                        Long updatedAmount = (currentAmount != null) ? currentAmount + prevRawUsage.getAmount() : prevRawUsage.getAmount();
+                        perRangeUnitToAmount.put(prevRawUsage.getUnitType(), updatedAmount);
+                        prevRawUsage = null;
+                    }
+                }
+
+                //
+                // If prevRawUsage != null it means that our first and current rawSubscriptionUsage does not fall into that interval; we can't
+                // just 'continue' as we need to correctly set next 'prevDate'
+                // If prevRawUsage == null, then consume as much as we can for that interval. Note that the stop condition requires consuming
+                // one additional element which will become the prevRawUsage for the next interval.
+                //
+                if (prevRawUsage == null) {
+                    while (rawUsageIterator.hasNext()) {
+                        final RawUsage curRawUsage = rawUsageIterator.next();
+                        if (curRawUsage.getDate().compareTo(curDate) >= 0) {
+                            prevRawUsage = curRawUsage;
+                            break;
+                        }
+
+                        Long currentAmount = perRangeUnitToAmount.get(curRawUsage.getUnitType());
+                        Long updatedAmount = (currentAmount != null) ? currentAmount + curRawUsage.getAmount() : curRawUsage.getAmount();
+                        perRangeUnitToAmount.put(curRawUsage.getUnitType(), updatedAmount);
+                    }
+                }
+
+                // If we did find some usage for that date range, let's populate the result
+                if (!perRangeUnitToAmount.isEmpty()) {
+                    final List<RolledUpUnit> rolledUpUnits = new ArrayList<RolledUpUnit>(perRangeUnitToAmount.size());
+                    for (final String unitType : perRangeUnitToAmount.keySet()) {
+                        rolledUpUnits.add(new DefaultRolledUpUnit(unitType, perRangeUnitToAmount.get(unitType)));
+                    }
+                    result.add(new DefaultRolledUpUsage(getSubscriptionId(), prevDate, curDate, rolledUpUnits));
+                }
+            }
+            prevDate = curDate;
+        }
+        return result;
     }
 
     /**

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/usage/DefaultRolledUpUnit.java b/invoice/src/main/java/org/killbill/billing/invoice/usage/DefaultRolledUpUnit.java
new file mode 100644
index 0000000..0e1976c
--- /dev/null
+++ b/invoice/src/main/java/org/killbill/billing/invoice/usage/DefaultRolledUpUnit.java
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.invoice.usage;
+
+import org.killbill.billing.usage.api.RolledUpUnit;
+
+public class DefaultRolledUpUnit implements RolledUpUnit {
+
+    private final String unitType;
+    private final Long amount;
+
+    public DefaultRolledUpUnit(final String unitType, final Long amount) {
+        this.unitType = unitType;
+        this.amount = amount;
+    }
+
+    @Override
+    public String getUnitType() {
+        return unitType;
+    }
+
+    @Override
+    public Long getAmount() {
+        return amount;
+    }
+}

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/usage/DefaultRolledUpUsage.java b/invoice/src/main/java/org/killbill/billing/invoice/usage/DefaultRolledUpUsage.java
new file mode 100644
index 0000000..df6e3d5
--- /dev/null
+++ b/invoice/src/main/java/org/killbill/billing/invoice/usage/DefaultRolledUpUsage.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.invoice.usage;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.joda.time.LocalDate;
+import org.killbill.billing.usage.api.RolledUpUnit;
+import org.killbill.billing.usage.api.RolledUpUsage;
+
+public class DefaultRolledUpUsage implements RolledUpUsage {
+
+    private final UUID subscriptionId;
+    private final LocalDate startDate;
+    private final LocalDate endDate;
+    private final List<RolledUpUnit> rolledUpUnits;
+
+    public DefaultRolledUpUsage(final UUID subscriptionId, final LocalDate startDate, final LocalDate endDate, final List<RolledUpUnit> rolledUpUnits) {
+        this.subscriptionId = subscriptionId;
+        this.startDate = startDate;
+        this.endDate = endDate;
+        this.rolledUpUnits = rolledUpUnits;
+    }
+
+    @Override
+    public UUID getSubscriptionId() {
+        return subscriptionId;
+    }
+
+    @Override
+    public LocalDate getStart() {
+        return startDate;
+    }
+
+    @Override
+    public LocalDate getEnd() {
+        return endDate;
+    }
+
+    @Override
+    public List<RolledUpUnit> getRolledUpUnits() {
+        return rolledUpUnits;
+    }
+}
+

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/usage/SubscriptionConsumableInArrear.java b/invoice/src/main/java/org/killbill/billing/invoice/usage/SubscriptionConsumableInArrear.java
index 8da11ed..9dfa26a 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/usage/SubscriptionConsumableInArrear.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/usage/SubscriptionConsumableInArrear.java
@@ -17,6 +17,7 @@
 package org.killbill.billing.invoice.usage;
 
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -31,33 +32,54 @@ import org.killbill.billing.catalog.api.Usage;
 import org.killbill.billing.catalog.api.UsageType;
 import org.killbill.billing.invoice.api.InvoiceItem;
 import org.killbill.billing.junction.BillingEvent;
-import org.killbill.billing.usage.api.UsageUserApi;
-import org.killbill.billing.util.callcontext.TenantContext;
+import org.killbill.billing.usage.RawUsage;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Function;
+import com.google.common.base.Predicate;
 import com.google.common.collect.Collections2;
+import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Ordering;
 
 /**
  * There is one such class created for each subscriptionId referenced in the billingEvents.
  */
 public class SubscriptionConsumableInArrear {
 
+    private static final Comparator<RawUsage> RAW_USAGE_DATE_COMPARATOR = new Comparator<RawUsage>() {
+        @Override
+        public int compare(final RawUsage o1, final RawUsage o2) {
+            int compared = o1.getDate().compareTo(o2.getDate());
+            if (compared != 0) {
+                return compared;
+            } else {
+                compared = o1.getUnitType().compareTo(o2.getUnitType());
+                if (compared != 0) {
+                    return compared;
+                } else {
+                    return o1.hashCode() != o2.hashCode() ? o1.hashCode() - o2.hashCode() : 0;
+                }
+            }
+        }
+    };
+
     private final UUID invoiceId;
     private final List<BillingEvent> subscriptionBillingEvents;
-    private final UsageUserApi usageApi;
     private final LocalDate targetDate;
-    private final TenantContext context;
-    private final boolean insertZeroAmountItems;
+    private final List<RawUsage> rawSubscriptionUsage;
 
-    public SubscriptionConsumableInArrear(final UUID invoiceId, final List<BillingEvent> subscriptionBillingEvents, final UsageUserApi usageApi, final boolean insertZeroAmountItems, LocalDate targetDate, final TenantContext context) {
+    public SubscriptionConsumableInArrear(final UUID invoiceId, final List<BillingEvent> subscriptionBillingEvents, final List<RawUsage> rawUsage, LocalDate targetDate) {
         this.invoiceId = invoiceId;
         this.subscriptionBillingEvents = subscriptionBillingEvents;
-        this.usageApi = usageApi;
-        this.insertZeroAmountItems = insertZeroAmountItems;
         this.targetDate = targetDate;
-        this.context = context;
+        // Extract raw usage for that subscription and sort it by date
+        this.rawSubscriptionUsage = Ordering.<RawUsage>from(RAW_USAGE_DATE_COMPARATOR).sortedCopy(Iterables.filter(rawUsage, new Predicate<RawUsage>() {
+            @Override
+            public boolean apply(final RawUsage input) {
+                return input.getSubscriptionId().equals(subscriptionBillingEvents.get(0).getSubscription().getId());
+            }
+        }));
     }
 
     /**
@@ -88,7 +110,6 @@ public class SubscriptionConsumableInArrear {
 
         for (BillingEvent event : subscriptionBillingEvents) {
 
-
             // Extract all in arrear /consumable usage section for that billing event.
             final List<Usage> usages = findConsumableInArrearUsages(event);
             allSeenUsage.addAll(Collections2.transform(usages, new Function<Usage, String>() {
@@ -106,7 +127,7 @@ public class SubscriptionConsumableInArrear {
                 // Add inflight usage interval if non existent
                 ContiguousIntervalConsumableInArrear existingInterval = inFlightInArrearUsageIntervals.get(usage.getName());
                 if (existingInterval == null) {
-                    existingInterval = new ContiguousIntervalConsumableInArrear(usage, invoiceId, usageApi, insertZeroAmountItems, targetDate, context);
+                    existingInterval = new ContiguousIntervalConsumableInArrear(usage, invoiceId, rawSubscriptionUsage, targetDate);
                     inFlightInArrearUsageIntervals.put(usage.getName(), existingInterval);
                 }
                 // Add billing event for that usage interval

diff --git a/invoice/src/test/java/org/killbill/billing/invoice/generator/TestDefaultInvoiceGenerator.java b/invoice/src/test/java/org/killbill/billing/invoice/generator/TestDefaultInvoiceGenerator.java
index b52e885..e83ad6f 100644
--- a/invoice/src/test/java/org/killbill/billing/invoice/generator/TestDefaultInvoiceGenerator.java
+++ b/invoice/src/test/java/org/killbill/billing/invoice/generator/TestDefaultInvoiceGenerator.java
@@ -119,16 +119,11 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
             }
 
             @Override
-            public boolean isInsertZeroUsageItems() {
-                return true;
-            }
-
-            @Override
             public TimeSpan getDryRunNotificationSchedule() {
                 return new TimeSpan("0s");
             }
         };
-        this.generator = new DefaultInvoiceGenerator(clock, null, invoiceConfig, internalCallContextFactory);
+        this.generator = new DefaultInvoiceGenerator(clock, null, invoiceConfig);
         this.account = new MockAccountBuilder().name(UUID.randomUUID().toString().substring(1, 8))
                                                .firstNameLength(6)
                                                .email(UUID.randomUUID().toString().substring(1, 8))

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 d73bcbe..5961ded 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
@@ -35,14 +35,17 @@ import org.killbill.billing.invoice.api.InvoiceItem;
 import org.killbill.billing.invoice.model.FixedPriceInvoiceItem;
 import org.killbill.billing.invoice.model.UsageInvoiceItem;
 import org.killbill.billing.junction.BillingEvent;
-import org.killbill.billing.usage.api.RolledUpUnit;
+import org.killbill.billing.usage.RawUsage;
 import org.killbill.billing.usage.api.RolledUpUsage;
-import org.killbill.billing.usage.api.user.DefaultRolledUpUsage;
+import org.killbill.billing.usage.api.svcs.DefaultRawUsage;
+import org.testng.Assert;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import com.google.common.base.Predicate;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 
 import static org.testng.Assert.assertEquals;
@@ -58,8 +61,6 @@ public class TestContiguousIntervalConsumableInArrear extends TestUsageInArrearB
     @BeforeMethod(groups = "fast")
     public void beforeMethod() {
         super.beforeMethod();
-        // Default invoice test binding;
-        this.mockUsageUserApi = usageUserApi;
     }
 
     @Test(groups = "fast")
@@ -73,7 +74,7 @@ public class TestContiguousIntervalConsumableInArrear extends TestUsageInArrearB
         final DefaultUsage usage = createDefaultUsage(usageName, tier);
 
         final LocalDate targetDate = startDate.plusDays(1);
-        final ContiguousIntervalConsumableInArrear intervalConsumableInArrear = createContiguousIntervalConsumableInArrear(usage, targetDate, false,
+        final ContiguousIntervalConsumableInArrear intervalConsumableInArrear = createContiguousIntervalConsumableInArrear(usage, ImmutableList.<RawUsage>of(), targetDate, false,
                                                                                                                            createMockBillingEvent(targetDate.toDateTimeAtStartOfDay(DateTimeZone.UTC),
                                                                                                                                                   Collections.<Usage>emptyList())
                                                                                                                           );
@@ -111,7 +112,7 @@ public class TestContiguousIntervalConsumableInArrear extends TestUsageInArrearB
         final DefaultUsage usage = createDefaultUsage(usageName, tier1, tier2);
 
         final LocalDate targetDate = new LocalDate(2014, 03, 20);
-        final ContiguousIntervalConsumableInArrear intervalConsumableInArrear = createContiguousIntervalConsumableInArrear(usage, targetDate, false,
+        final ContiguousIntervalConsumableInArrear intervalConsumableInArrear = createContiguousIntervalConsumableInArrear(usage, ImmutableList.<RawUsage>of(), targetDate, false,
                                                                                                                            createMockBillingEvent(targetDate.toDateTimeAtStartOfDay(DateTimeZone.UTC),
                                                                                                                                                   Collections.<Usage>emptyList())
                                                                                                                           );
@@ -130,18 +131,11 @@ public class TestContiguousIntervalConsumableInArrear extends TestUsageInArrearB
         final LocalDate endDate = new LocalDate(2014, 05, 15);
 
         // 2 items for startDate - firstBCDDate
-        final List<RolledUpUnit> units1 = new ArrayList<RolledUpUnit>();
-        units1.add(createRolledUpUnit("unit", 130L));
-        units1.add(createRolledUpUnit("unit", 271L));
-        final RolledUpUsage usage1 = new DefaultRolledUpUsage(subscriptionId, startDate, firstBCDDate, units1);
-
+        final List<RawUsage> rawUsages = new ArrayList<RawUsage>();
+        rawUsages.add(new DefaultRawUsage(subscriptionId, new LocalDate(2014, 03, 20), "unit", 130L));
+        rawUsages.add(new DefaultRawUsage(subscriptionId, new LocalDate(2014, 03, 21), "unit", 271L));
         // 1 items for firstBCDDate - endDate
-        final List<RolledUpUnit> units2 = new ArrayList<RolledUpUnit>();
-        units2.add(createRolledUpUnit("unit", 199L));
-        final RolledUpUsage usage3 = new DefaultRolledUpUsage(subscriptionId, firstBCDDate, endDate, units2);
-
-        final List<RolledUpUsage> usages = ImmutableList.<RolledUpUsage>builder().add(usage1).add(usage3).build();
-        this.mockUsageUserApi = createMockUsageUserApi(usages);
+        rawUsages.add(new DefaultRawUsage(subscriptionId, new LocalDate(2014, 04, 15), "unit", 199L));
 
         final DefaultTieredBlock block = createDefaultTieredBlock("unit", 100, 10, BigDecimal.ONE);
         final DefaultTier tier = createDefaultTier(block);
@@ -152,7 +146,7 @@ public class TestContiguousIntervalConsumableInArrear extends TestUsageInArrearB
         final BillingEvent event1 = createMockBillingEvent(startDate.toDateTimeAtStartOfDay(DateTimeZone.UTC), Collections.<Usage>emptyList());
         final BillingEvent event2 = createMockBillingEvent(endDate.toDateTimeAtStartOfDay(DateTimeZone.UTC), Collections.<Usage>emptyList());
 
-        final ContiguousIntervalConsumableInArrear intervalConsumableInArrear = createContiguousIntervalConsumableInArrear(usage, targetDate, true, event1, event2);
+        final ContiguousIntervalConsumableInArrear intervalConsumableInArrear = createContiguousIntervalConsumableInArrear(usage, rawUsages, targetDate, true, event1, event2);
 
         final List<InvoiceItem> invoiceItems = new ArrayList<InvoiceItem>();
         final InvoiceItem ii1 = new UsageInvoiceItem(invoiceId, accountId, bundleId, subscriptionId, planName, phaseName, usage.getName(), startDate, firstBCDDate, BigDecimal.ONE, currency);
@@ -161,8 +155,17 @@ public class TestContiguousIntervalConsumableInArrear extends TestUsageInArrearB
         final InvoiceItem ii2 = new UsageInvoiceItem(invoiceId, accountId, bundleId, subscriptionId, planName, phaseName, usage.getName(), firstBCDDate, endDate, BigDecimal.ONE, currency);
         invoiceItems.add(ii2);
 
-        final List<InvoiceItem> result = intervalConsumableInArrear.computeMissingItems(invoiceItems);
-        assertEquals(result.size(), 2);
+        final List<InvoiceItem> rawResults = intervalConsumableInArrear.computeMissingItems(invoiceItems);
+        assertEquals(rawResults.size(), 4);
+
+        final List<InvoiceItem> result = ImmutableList.copyOf(Iterables.filter(rawResults, new Predicate<InvoiceItem>() {
+            @Override
+            public boolean apply(final InvoiceItem input) {
+                return input.getAmount().compareTo(BigDecimal.ZERO) > 0;
+            }
+        }));
+
+
         // Invoiced for 1 BTC and used 130 + 271 = 401 => 5 blocks => 5 BTC so remaining piece should be 4 BTC
         assertEquals(result.get(0).getAmount().compareTo(new BigDecimal("4.0")), 0, String.format("%s != 4.0", result.get(0).getAmount()));
         assertEquals(result.get(0).getCurrency(), Currency.BTC);
@@ -188,16 +191,69 @@ public class TestContiguousIntervalConsumableInArrear extends TestUsageInArrearB
         assertTrue(result.get(1).getEndDate().compareTo(endDate) == 0);
     }
 
-    private RolledUpUnit createRolledUpUnit(final String unit, final Long amount) {
-        return new RolledUpUnit() {
-            @Override
-            public String getUnitType() {
-                return unit;
-            }
-            @Override
-            public Long getAmount() {
-                return amount;
-            }
-        };
+    @Test(groups = "fast")
+    public void testGetRolledUpUsage() {
+
+        final DefaultTieredBlock tieredBlock1 = createDefaultTieredBlock("unit", 100, 1000, BigDecimal.ONE);
+        final DefaultTieredBlock tieredBlock2 = createDefaultTieredBlock("other_unit", 10, 1000, BigDecimal.ONE);
+        final DefaultTier tier = createDefaultTier(tieredBlock1, tieredBlock2);
+
+
+        final DefaultUsage usage = createDefaultUsage(usageName, tier);
+
+
+        final LocalDate t0 = new LocalDate(2015, 03, BCD);
+        final BillingEvent eventT0 = createMockBillingEvent(t0.toDateTimeAtStartOfDay(DateTimeZone.UTC), Collections.<Usage>emptyList());
+
+        final LocalDate t1 = new LocalDate(2015, 04, BCD);
+        final BillingEvent eventT1 = createMockBillingEvent(t1.toDateTimeAtStartOfDay(DateTimeZone.UTC), Collections.<Usage>emptyList());
+
+        final LocalDate t2 = new LocalDate(2015, 05, BCD);
+        final BillingEvent eventT2 = createMockBillingEvent(t2.toDateTimeAtStartOfDay(DateTimeZone.UTC), Collections.<Usage>emptyList());
+
+        final LocalDate t3 = new LocalDate(2015, 06, BCD);
+        final BillingEvent eventT3 = createMockBillingEvent(t3.toDateTimeAtStartOfDay(DateTimeZone.UTC), Collections.<Usage>emptyList());
+
+        final LocalDate targetDate = t3;
+
+
+        // Prev t0
+        final RawUsage raw1 = new DefaultRawUsage(subscriptionId, new LocalDate(2015, 03, 01), "unit", 12L);
+
+        // t0 - t1
+        final RawUsage raw2 = new DefaultRawUsage(subscriptionId, new LocalDate(2015, 03, 15), "unit", 6L);
+        final RawUsage raw3 = new DefaultRawUsage(subscriptionId, new LocalDate(2015, 03, 25), "unit", 4L);
+
+        // t1 - t2 nothing
+
+        // t2 - t3
+        final RawUsage raw4 = new DefaultRawUsage(subscriptionId, new LocalDate(2015, 05, 15), "unit", 13L);
+        final RawUsage oraw1 = new DefaultRawUsage(subscriptionId, new LocalDate(2015, 05, 21), "other_unit", 21L);
+        final RawUsage raw5 = new DefaultRawUsage(subscriptionId, new LocalDate(2015, 05, 31), "unit", 7L);
+
+        // after t3
+        final RawUsage raw6 = new DefaultRawUsage(subscriptionId, new LocalDate(2015, 06, 15), "unit", 100L);
+
+        final List<RawUsage> rawUsage = ImmutableList.of(raw1, raw2, raw3, raw4, oraw1, raw5, raw6);
+
+        final ContiguousIntervalConsumableInArrear intervalConsumableInArrear = createContiguousIntervalConsumableInArrear(usage, rawUsage, targetDate, true, eventT0, eventT1, eventT2, eventT3);
+
+
+        final List<RolledUpUsage> rolledUpUsage =  intervalConsumableInArrear.getRolledUpUsage();
+        Assert.assertEquals(rolledUpUsage.size(), 2);
+
+        Assert.assertEquals(rolledUpUsage.get(0).getStart().compareTo(t0), 0);
+        Assert.assertEquals(rolledUpUsage.get(0).getEnd().compareTo(t1), 0);
+        Assert.assertEquals(rolledUpUsage.get(0).getRolledUpUnits().size(),1);
+        Assert.assertEquals(rolledUpUsage.get(0).getRolledUpUnits().get(0).getUnitType(), "unit");
+        Assert.assertEquals(rolledUpUsage.get(0).getRolledUpUnits().get(0).getAmount(), new Long(10L));
+
+        Assert.assertEquals(rolledUpUsage.get(1).getStart().compareTo(t2), 0);
+        Assert.assertEquals(rolledUpUsage.get(1).getEnd().compareTo(t3), 0);
+        Assert.assertEquals(rolledUpUsage.get(1).getRolledUpUnits().size(),2);
+        Assert.assertEquals(rolledUpUsage.get(1).getRolledUpUnits().get(0).getUnitType(), "unit");
+        Assert.assertEquals(rolledUpUsage.get(1).getRolledUpUnits().get(0).getAmount(), new Long(20L));
+        Assert.assertEquals(rolledUpUsage.get(1).getRolledUpUnits().get(1).getUnitType(), "other_unit");
+        Assert.assertEquals(rolledUpUsage.get(1).getRolledUpUnits().get(1).getAmount(), new Long(21L));
     }
 }

diff --git a/invoice/src/test/java/org/killbill/billing/invoice/usage/TestSubscriptionConsumableInArrear.java b/invoice/src/test/java/org/killbill/billing/invoice/usage/TestSubscriptionConsumableInArrear.java
index 9ab1951..90c20d6 100644
--- a/invoice/src/test/java/org/killbill/billing/invoice/usage/TestSubscriptionConsumableInArrear.java
+++ b/invoice/src/test/java/org/killbill/billing/invoice/usage/TestSubscriptionConsumableInArrear.java
@@ -26,6 +26,7 @@ import org.killbill.billing.catalog.DefaultTier;
 import org.killbill.billing.catalog.DefaultTieredBlock;
 import org.killbill.billing.catalog.api.Usage;
 import org.killbill.billing.junction.BillingEvent;
+import org.killbill.billing.usage.RawUsage;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
@@ -71,7 +72,7 @@ public class TestSubscriptionConsumableInArrear extends TestUsageInArrearBase {
 
         LocalDate targetDate = new LocalDate(2013, 6, 23);
 
-        final SubscriptionConsumableInArrear foo = new SubscriptionConsumableInArrear(invoiceId, billingEvents, usageUserApi, true, targetDate, callContext);
+        final SubscriptionConsumableInArrear foo = new SubscriptionConsumableInArrear(invoiceId, billingEvents, ImmutableList.<RawUsage>of(), targetDate);
         final List<ContiguousIntervalConsumableInArrear> result = foo.computeInArrearUsageInterval();
         assertEquals(result.size(), 3);
 
@@ -91,7 +92,5 @@ public class TestSubscriptionConsumableInArrear extends TestUsageInArrearBase {
         assertEquals(result.get(2).getTransitionTimes().size(), 2);
         assertTrue(result.get(2).getTransitionTimes().get(0).compareTo(new LocalDate(2013, 5, 23)) == 0);
         assertTrue(result.get(2).getTransitionTimes().get(1).compareTo(new LocalDate(2013, 6, 15)) == 0);
-
     }
-
 }

diff --git a/invoice/src/test/java/org/killbill/billing/invoice/usage/TestUsageInArrearBase.java b/invoice/src/test/java/org/killbill/billing/invoice/usage/TestUsageInArrearBase.java
index 6d66b40..f817cfb 100644
--- a/invoice/src/test/java/org/killbill/billing/invoice/usage/TestUsageInArrearBase.java
+++ b/invoice/src/test/java/org/killbill/billing/invoice/usage/TestUsageInArrearBase.java
@@ -18,7 +18,6 @@ package org.killbill.billing.invoice.usage;
 
 import java.math.BigDecimal;
 import java.util.List;
-import java.util.Set;
 import java.util.UUID;
 
 import org.joda.time.DateTime;
@@ -41,9 +40,8 @@ import org.killbill.billing.catalog.api.UsageType;
 import org.killbill.billing.invoice.InvoiceTestSuiteNoDB;
 import org.killbill.billing.junction.BillingEvent;
 import org.killbill.billing.subscription.api.SubscriptionBase;
-import org.killbill.billing.usage.api.RolledUpUsage;
+import org.killbill.billing.usage.RawUsage;
 import org.killbill.billing.usage.api.UsageUserApi;
-import org.killbill.billing.util.callcontext.TenantContext;
 import org.mockito.Mockito;
 import org.testng.annotations.BeforeClass;
 
@@ -61,7 +59,6 @@ public abstract class TestUsageInArrearBase extends InvoiceTestSuiteNoDB {
 
     protected UsageUserApi mockUsageUserApi;
 
-
     @BeforeClass(groups = "fast")
     protected void beforeClass() throws Exception {
         super.beforeClass();
@@ -76,14 +73,8 @@ public abstract class TestUsageInArrearBase extends InvoiceTestSuiteNoDB {
         currency = Currency.BTC;
     }
 
-    protected UsageUserApi createMockUsageUserApi(final List<RolledUpUsage> returnValue) {
-        final UsageUserApi result = Mockito.mock(UsageUserApi.class);
-        Mockito.when(result.getAllUsageForSubscription(Mockito.<UUID>any(), Mockito.<List<LocalDate>>any(), Mockito.<TenantContext>any())).thenReturn(returnValue);
-        return result;
-    }
-
-    protected ContiguousIntervalConsumableInArrear createContiguousIntervalConsumableInArrear(final DefaultUsage usage, final LocalDate targetDate, final boolean closedInterval, final BillingEvent... events) {
-        final ContiguousIntervalConsumableInArrear intervalConsumableInArrear = new ContiguousIntervalConsumableInArrear(usage, invoiceId, mockUsageUserApi, true, targetDate, callContext);
+    protected ContiguousIntervalConsumableInArrear createContiguousIntervalConsumableInArrear(final DefaultUsage usage, List<RawUsage> rawUsages, final LocalDate targetDate, final boolean closedInterval, final BillingEvent... events) {
+        final ContiguousIntervalConsumableInArrear intervalConsumableInArrear = new ContiguousIntervalConsumableInArrear(usage, invoiceId, rawUsages, targetDate);
         for (BillingEvent event : events) {
             intervalConsumableInArrear.addBillingEvent(event);
         }

diff --git a/usage/src/main/java/org/killbill/billing/usage/api/svcs/DefaultInternalUserApi.java b/usage/src/main/java/org/killbill/billing/usage/api/svcs/DefaultInternalUserApi.java
new file mode 100644
index 0000000..fc20656
--- /dev/null
+++ b/usage/src/main/java/org/killbill/billing/usage/api/svcs/DefaultInternalUserApi.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.usage.api.svcs;
+
+import java.util.List;
+import java.util.UUID;
+
+import javax.annotation.Nullable;
+import javax.inject.Inject;
+
+import org.joda.time.LocalDate;
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.usage.InternalUserApi;
+import org.killbill.billing.usage.RawUsage;
+import org.killbill.billing.usage.dao.RolledUpUsageDao;
+import org.killbill.billing.usage.dao.RolledUpUsageModelDao;
+
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+
+public class DefaultInternalUserApi implements InternalUserApi {
+
+    private final RolledUpUsageDao rolledUpUsageDao;
+
+    @Inject
+    public DefaultInternalUserApi(final RolledUpUsageDao rolledUpUsageDao) {
+        this.rolledUpUsageDao = rolledUpUsageDao;
+    }
+
+    @Override
+    public List<RawUsage> getRawUsageForAccount(final UUID accountId, final LocalDate stateDate, final LocalDate endDate, final InternalTenantContext internalTenantContext) {
+        final List<RolledUpUsageModelDao> usage = rolledUpUsageDao.getRawUsageForAccount(stateDate, endDate, internalTenantContext);
+        return ImmutableList.copyOf(Iterables.transform(usage, new Function<RolledUpUsageModelDao, RawUsage>() {
+            @Nullable
+            @Override
+            public RawUsage apply(final RolledUpUsageModelDao input) {
+                return new DefaultRawUsage(input.getSubscriptionId(), input.getRecordDate(), input.getUnitType(), input.getAmount());
+            }
+        }));
+    }
+}

diff --git a/usage/src/main/java/org/killbill/billing/usage/api/svcs/DefaultRawUsage.java b/usage/src/main/java/org/killbill/billing/usage/api/svcs/DefaultRawUsage.java
new file mode 100644
index 0000000..7ce8718
--- /dev/null
+++ b/usage/src/main/java/org/killbill/billing/usage/api/svcs/DefaultRawUsage.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.usage.api.svcs;
+
+import java.util.UUID;
+
+import org.joda.time.LocalDate;
+import org.killbill.billing.usage.RawUsage;
+
+public class DefaultRawUsage implements RawUsage {
+
+    private final UUID subscriptionId;
+    private final LocalDate recordDate;
+    private final String unitType;
+    private final Long amount;
+
+    public DefaultRawUsage(final UUID subscriptionId, final LocalDate recordDate, final String unitType, final Long amount) {
+        this.subscriptionId = subscriptionId;
+        this.recordDate = recordDate;
+        this.unitType = unitType;
+        this.amount = amount;
+    }
+
+    @Override
+    public UUID getSubscriptionId() {
+        return subscriptionId;
+    }
+
+    @Override
+    public LocalDate getDate() {
+        return recordDate;
+    }
+
+    @Override
+    public String getUnitType() {
+        return unitType;
+    }
+
+    @Override
+    public Long getAmount() {
+        return amount;
+    }
+}

diff --git a/usage/src/main/java/org/killbill/billing/usage/dao/DefaultRolledUpUsageDao.java b/usage/src/main/java/org/killbill/billing/usage/dao/DefaultRolledUpUsageDao.java
index 4923592..3ab2973 100644
--- a/usage/src/main/java/org/killbill/billing/usage/dao/DefaultRolledUpUsageDao.java
+++ b/usage/src/main/java/org/killbill/billing/usage/dao/DefaultRolledUpUsageDao.java
@@ -51,4 +51,9 @@ public class DefaultRolledUpUsageDao implements RolledUpUsageDao {
     public List<RolledUpUsageModelDao> getAllUsageForSubscription(final UUID subscriptionId, final LocalDate startDate, final LocalDate endDate, final InternalTenantContext context) {
         return rolledUpUsageSqlDao.getAllUsageForSubscription(subscriptionId, startDate.toDate(), endDate.toDate(), context);
     }
+
+    @Override
+    public List<RolledUpUsageModelDao> getRawUsageForAccount(final LocalDate startDate, final LocalDate endDate, final InternalTenantContext context) {
+        return rolledUpUsageSqlDao.getRawUsageForAccount(startDate.toDate(), endDate.toDate(), context);
+    }
 }

diff --git a/usage/src/main/java/org/killbill/billing/usage/dao/RolledUpUsageDao.java b/usage/src/main/java/org/killbill/billing/usage/dao/RolledUpUsageDao.java
index 4144aed..e458635 100644
--- a/usage/src/main/java/org/killbill/billing/usage/dao/RolledUpUsageDao.java
+++ b/usage/src/main/java/org/killbill/billing/usage/dao/RolledUpUsageDao.java
@@ -31,4 +31,7 @@ public interface RolledUpUsageDao {
     List<RolledUpUsageModelDao> getUsageForSubscription(UUID subscriptionId, LocalDate startDate, LocalDate endDate, String unitType, InternalTenantContext context);
 
     List<RolledUpUsageModelDao> getAllUsageForSubscription(UUID subscriptionId, LocalDate startDate, LocalDate endDate, InternalTenantContext context);
+
+
+    List<RolledUpUsageModelDao> getRawUsageForAccount(LocalDate startDate, LocalDate endDate, InternalTenantContext context);
 }

diff --git a/usage/src/main/java/org/killbill/billing/usage/dao/RolledUpUsageSqlDao.java b/usage/src/main/java/org/killbill/billing/usage/dao/RolledUpUsageSqlDao.java
index b237eb2..f921988 100644
--- a/usage/src/main/java/org/killbill/billing/usage/dao/RolledUpUsageSqlDao.java
+++ b/usage/src/main/java/org/killbill/billing/usage/dao/RolledUpUsageSqlDao.java
@@ -50,4 +50,9 @@ public interface RolledUpUsageSqlDao extends EntitySqlDao<RolledUpUsageModelDao,
                                                                   @Bind("startDate") final Date startDate,
                                                                   @Bind("endDate") final Date endDate,
                                                                   @InternalTenantContextBinder final InternalTenantContext context);
+
+    @SqlQuery
+    public List<RolledUpUsageModelDao> getRawUsageForAccount(@Bind("startDate") final Date startDate,
+                                                             @Bind("endDate") final Date endDate,
+                                                             @InternalTenantContextBinder final InternalTenantContext context);
 }

diff --git a/usage/src/main/java/org/killbill/billing/usage/glue/UsageModule.java b/usage/src/main/java/org/killbill/billing/usage/glue/UsageModule.java
index 749eba1..f4c83dd 100644
--- a/usage/src/main/java/org/killbill/billing/usage/glue/UsageModule.java
+++ b/usage/src/main/java/org/killbill/billing/usage/glue/UsageModule.java
@@ -19,7 +19,9 @@
 package org.killbill.billing.usage.glue;
 
 import org.killbill.billing.platform.api.KillbillConfigSource;
+import org.killbill.billing.usage.InternalUserApi;
 import org.killbill.billing.usage.api.UsageUserApi;
+import org.killbill.billing.usage.api.svcs.DefaultInternalUserApi;
 import org.killbill.billing.usage.api.user.DefaultUsageUserApi;
 import org.killbill.billing.usage.dao.DefaultRolledUpUsageDao;
 import org.killbill.billing.usage.dao.RolledUpUsageDao;
@@ -39,9 +41,15 @@ public class UsageModule extends KillBillModule {
         bind(UsageUserApi.class).to(DefaultUsageUserApi.class).asEagerSingleton();
     }
 
+    protected void installInternalUserApi() {
+        bind(InternalUserApi.class).to(DefaultInternalUserApi.class).asEagerSingleton();
+    }
+
+
     @Override
     protected void configure() {
         installRolledUpUsageDao();
         installUsageUserApi();
+        installInternalUserApi();
     }
 }

diff --git a/usage/src/main/resources/org/killbill/billing/usage/dao/RolledUpUsageSqlDao.sql.stg b/usage/src/main/resources/org/killbill/billing/usage/dao/RolledUpUsageSqlDao.sql.stg
index ea40a86..7672594 100644
--- a/usage/src/main/resources/org/killbill/billing/usage/dao/RolledUpUsageSqlDao.sql.stg
+++ b/usage/src/main/resources/org/killbill/billing/usage/dao/RolledUpUsageSqlDao.sql.stg
@@ -45,3 +45,15 @@ and record_date \< :endDate
 ;
 >>
 
+getRawUsageForAccount() ::= <<
+select
+  <allTableFields()>
+from <tableName()>
+where account_record_id = :accountRecordId
+and record_date >= :startDate
+and record_date \< :endDate
+<AND_CHECK_TENANT()>
+;
+>>
+
+

diff --git a/usage/src/main/resources/org/killbill/billing/usage/ddl.sql b/usage/src/main/resources/org/killbill/billing/usage/ddl.sql
index 50b3f2d..3974dab 100644
--- a/usage/src/main/resources/org/killbill/billing/usage/ddl.sql
+++ b/usage/src/main/resources/org/killbill/billing/usage/ddl.sql
@@ -17,3 +17,4 @@ CREATE TABLE rolled_up_usage (
 CREATE UNIQUE INDEX rolled_up_usage_id ON rolled_up_usage(id);
 CREATE INDEX rolled_up_usage_subscription_id ON rolled_up_usage(subscription_id ASC);
 CREATE INDEX rolled_up_usage_tenant_account_record_id ON rolled_up_usage(tenant_record_id, account_record_id);
+CREATE INDEX rolled_up_usage_account_record_id ON rolled_up_usage(account_record_id);

diff --git a/usage/src/test/java/org/killbill/billing/usage/api/user/MockUsageUserApi.java b/usage/src/test/java/org/killbill/billing/usage/api/user/MockUsageUserApi.java
index 4ec7e95..3d6d5ed 100644
--- a/usage/src/test/java/org/killbill/billing/usage/api/user/MockUsageUserApi.java
+++ b/usage/src/test/java/org/killbill/billing/usage/api/user/MockUsageUserApi.java
@@ -17,7 +17,6 @@
 package org.killbill.billing.usage.api.user;
 
 import java.util.List;
-import java.util.Set;
 import java.util.UUID;
 
 import org.joda.time.LocalDate;
@@ -49,5 +48,4 @@ public class MockUsageUserApi implements UsageUserApi {
     public List<RolledUpUsage> getAllUsageForSubscription(final UUID uuid, final List<LocalDate> localDates, final TenantContext tenantContext) {
         return null;
     }
-
 }

diff --git a/util/src/main/java/org/killbill/billing/util/config/InvoiceConfig.java b/util/src/main/java/org/killbill/billing/util/config/InvoiceConfig.java
index 6d5a205..de5b308 100644
--- a/util/src/main/java/org/killbill/billing/util/config/InvoiceConfig.java
+++ b/util/src/main/java/org/killbill/billing/util/config/InvoiceConfig.java
@@ -33,11 +33,6 @@ public interface InvoiceConfig extends KillbillConfig {
     @Description("Whether to send email notifications on invoice creation (for configured accounts)")
     public boolean isEmailNotificationsEnabled();
 
-    @Config("org.killbill.invoice.usage.insert.zero.amount")
-    @Default("true")
-    @Description("Whether to insert usage items with a zero amount")
-    public boolean isInsertZeroUsageItems();
-
     @Config("org.killbill.invoice.dryRunNotificationSchedule")
     @Default("0s")
     @Description("DryRun invoice notification time before targetDate (ignored if set to 0s)")