killbill-aplcache

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 dcb4b2a..5ecadbf 100644
--- a/api/src/main/java/org/killbill/billing/junction/BillingEventSet.java
+++ b/api/src/main/java/org/killbill/billing/junction/BillingEventSet.java
@@ -32,6 +32,8 @@ public interface BillingEventSet extends SortedSet<BillingEvent> {
 
     public boolean isAccountAutoInvoiceDraft();
 
+    public boolean isAccountAutoInvoiceReuseDraft();
+
     public BillingMode getRecurringBillingMode();
 
     public List<UUID> getSubscriptionIdsWithAutoInvoiceOff();

diff --git a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationBase.java b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationBase.java
index d38cea6..87e06b6 100644
--- a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationBase.java
+++ b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationBase.java
@@ -810,12 +810,15 @@ public class TestIntegrationBase extends BeatrixTestSuiteWithEmbeddedDB {
         add_account_Tag(id, ControlTagType.AUTO_INVOICING_DRAFT, type);
     }
 
+    protected void add_AUTO_INVOICING_REUSE_DRAFT_Tag(final UUID id, final ObjectType type) throws TagDefinitionApiException, TagApiException {
+        add_account_Tag(id, ControlTagType.AUTO_INVOICING_REUSE_DRAFT, type);
+    }
+
     private void add_account_Tag(final UUID id, final ControlTagType controlTagType, final ObjectType type) throws TagDefinitionApiException, TagApiException {
         busHandler.pushExpectedEvent(NextEvent.TAG);
         tagUserApi.addTag(id, type, controlTagType.getId(), callContext);
         assertListenerStatus();
-        final List<Tag> tags = tagUserApi.getTagsForObject(id, type, false, callContext);
-        assertEquals(tags.size(), 1);
+        tagUserApi.getTagsForObject(id, type, false, callContext);
     }
 
     protected void remove_AUTO_PAY_OFF_Tag(final UUID id, final ObjectType type, final NextEvent... additionalEvents) throws TagDefinitionApiException, TagApiException {

diff --git a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationWithAutoInvoiceDraft.java b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationWithAutoInvoiceDraft.java
index 5353676..3956312 100644
--- a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationWithAutoInvoiceDraft.java
+++ b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestIntegrationWithAutoInvoiceDraft.java
@@ -30,12 +30,15 @@ import org.killbill.billing.account.api.Account;
 import org.killbill.billing.api.TestApiListener.NextEvent;
 import org.killbill.billing.beatrix.util.InvoiceChecker.ExpectedInvoiceItemCheck;
 import org.killbill.billing.catalog.api.BillingPeriod;
+import org.killbill.billing.catalog.api.Currency;
 import org.killbill.billing.catalog.api.ProductCategory;
 import org.killbill.billing.entitlement.api.DefaultEntitlement;
 import org.killbill.billing.invoice.api.Invoice;
+import org.killbill.billing.invoice.api.InvoiceItem;
 import org.killbill.billing.invoice.api.InvoiceItemType;
 import org.killbill.billing.invoice.api.InvoiceStatus;
 import org.killbill.billing.invoice.api.InvoiceUserApi;
+import org.killbill.billing.invoice.model.ExternalChargeInvoiceItem;
 import org.killbill.billing.util.api.TagApiException;
 import org.killbill.billing.util.api.TagDefinitionApiException;
 import org.killbill.billing.util.api.TagUserApi;
@@ -74,7 +77,7 @@ public class TestIntegrationWithAutoInvoiceDraft extends TestIntegrationBase {
 
 
     @Test(groups = "slow")
-    public void testBasic() throws Exception {
+    public void testAutoInvoicingDraftBasic() throws Exception {
         clock.setTime(new DateTime(2017, 6, 16, 18, 24, 42, 0));
         add_AUTO_INVOICING_DRAFT_Tag(account.getId(), ObjectType.ACCOUNT);
 
@@ -195,4 +198,77 @@ public class TestIntegrationWithAutoInvoiceDraft extends TestIntegrationBase {
 
     }
 
+
+    @Test(groups = "slow")
+    public void testAutoInvoicingReuseDraftBasic() throws Exception {
+        clock.setTime(new DateTime(2017, 6, 16, 18, 24, 42, 0));
+
+        // Set both AUTO_INVOICING_DRAFT and AUTO_INVOICING_REUSE_DRAFT
+        add_AUTO_INVOICING_DRAFT_Tag(account.getId(), ObjectType.ACCOUNT);
+        add_AUTO_INVOICING_REUSE_DRAFT_Tag(account.getId(), ObjectType.ACCOUNT);
+
+        // Create initial DRAFt invoice that will be reused by the system
+        final LocalDate startDate = clock.getUTCToday();
+        final LocalDate endDate = startDate.plusDays(5);
+        final InvoiceItem externalCharge = new ExternalChargeInvoiceItem(null, account.getId(), null, "Initial external charge", startDate, endDate, BigDecimal.TEN, Currency.USD);
+        invoiceUserApi.insertExternalCharges(account.getId(), clock.getUTCToday(), ImmutableList.<InvoiceItem>of(externalCharge), false, callContext).get(0);
+
+        List<Invoice> invoices;
+        invoices = invoiceApi.getInvoicesByAccount(account.getId(), false, callContext);
+        assertEquals(invoices.size(), 1);
+        assertEquals(invoices.get(0).getInvoiceItems().size(), 1);
+        assertEquals(invoices.get(0).getStatus(), InvoiceStatus.DRAFT);
+
+        final UUID invoiceId = invoices.get(0).getId();
+
+        final DefaultEntitlement bpEntitlement = createBaseEntitlementAndCheckForCompletion(account.getId(), "externalKey", productName, ProductCategory.BASE, term, NextEvent.CREATE, NextEvent.BLOCK);
+        assertNotNull(bpEntitlement);
+
+        // Verify we see the new item on our existing DRAFT invoice
+        invoices = invoiceApi.getInvoicesByAccount(account.getId(), false, callContext);
+        assertEquals(invoices.size(), 1);
+        assertEquals(invoices.get(0).getId(), invoiceId);
+        assertEquals(invoices.get(0).getInvoiceItems().size(), 2);
+        assertEquals(invoices.get(0).getStatus(), InvoiceStatus.DRAFT);
+
+
+        // Move out of TRIAL
+        busHandler.pushExpectedEvents(NextEvent.PHASE);
+        clock.addDays(30);
+        assertListenerStatus();
+
+        // Verify again we see the new item on our existing DRAFT invoice
+        invoices = invoiceApi.getInvoicesByAccount(account.getId(), false, callContext);
+        assertEquals(invoices.size(), 1);
+        assertEquals(invoices.get(0).getId(), invoiceId);
+        assertEquals(invoices.get(0).getInvoiceItems().size(), 3);
+        assertEquals(invoices.get(0).getStatus(), InvoiceStatus.DRAFT);
+
+        // Remove AUTO_INVOICING_DRAFT, so next invoicing should commit DRAFt invoice
+        remove_AUTO_INVOICING_DRAFT_Tag(account.getId(), ObjectType.ACCOUNT);
+
+        busHandler.pushExpectedEvents(NextEvent.INVOICE, NextEvent.PAYMENT, NextEvent.INVOICE_PAYMENT);
+        clock.addMonths(1);
+        assertListenerStatus();
+
+        // Verify again we see the new item and this time invoice is in COMMITTED
+        invoices = invoiceApi.getInvoicesByAccount(account.getId(), false, callContext);
+        assertEquals(invoices.size(), 1);
+        assertEquals(invoices.get(0).getId(), invoiceId);
+        assertEquals(invoices.get(0).getInvoiceItems().size(), 4);
+        assertEquals(invoices.get(0).getStatus(), InvoiceStatus.COMMITTED);
+
+
+        // Verify we see a new invoice
+        busHandler.pushExpectedEvents(NextEvent.INVOICE, NextEvent.PAYMENT, NextEvent.INVOICE_PAYMENT);
+        clock.addMonths(1);
+        assertListenerStatus();
+
+        invoices = invoiceApi.getInvoicesByAccount(account.getId(), false, callContext);
+        assertEquals(invoices.size(), 2);
+        assertEquals(invoices.get(1).getInvoiceItems().size(), 1);
+        assertEquals(invoices.get(1).getStatus(), InvoiceStatus.COMMITTED);
+    }
+
+
 }

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 5537dee..568d84b 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
@@ -276,6 +276,17 @@ public class DefaultInvoiceDao extends EntityDaoBase<InvoiceModelDao, Invoice, I
     }
 
     @Override
+    public InvoiceModelDao getEarliestDraftInvoiceByAccount(final InternalTenantContext context) {
+        return transactionalSqlDao.execute(new EntitySqlDaoTransactionWrapper<InvoiceModelDao>() {
+            @Override
+            public InvoiceModelDao inTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory) throws Exception {
+                final InvoiceSqlDao invoiceDao = entitySqlDaoWrapperFactory.become(InvoiceSqlDao.class);
+                return invoiceDao.getEarliestDraftInvoiceByAccount(context);
+            }
+        });
+    }
+
+    @Override
     public void createInvoice(final InvoiceModelDao invoice,
                               final FutureAccountNotifications callbackDateTimePerSubscriptions,
                               final InternalCallContext context) {
@@ -291,9 +302,14 @@ public class DefaultInvoiceDao extends EntityDaoBase<InvoiceModelDao, Invoice, I
     private List<InvoiceItemModelDao> createInvoices(final Iterable<InvoiceModelDao> invoices,
                                                      final FutureAccountNotifications callbackDateTimePerSubscriptions,
                                                      final InternalCallContext context) {
+        // Track invoices that are being created
         final Collection<UUID> createdInvoiceIds = new HashSet<UUID>();
-        final Collection<UUID> modifiedInvoiceIds = new HashSet<UUID>();
-        final Collection<UUID> committedInvoiceIds = new HashSet<UUID>();
+        // Track invoices that already exist but are being committed -- AUTO_INVOICING_REUSE_DRAFT mode
+        final Collection<UUID> committedReusedInvoiceId = new HashSet<UUID>();
+        // Track all invoices that are referenced through all invoiceItems
+        final Collection<UUID> allInvoiceIds = new HashSet<UUID>();
+        // Track invoices that are committed but were not created or reused -- to sent the InvoiceAdjustment bus event
+        final Collection<UUID> adjustedCommittedInvoiceIds = new HashSet<UUID>();
 
         final Collection<UUID> invoiceIdsReferencedFromItems = new HashSet<UUID>();
         for (final InvoiceModelDao invoiceModelDao : invoices) {
@@ -321,28 +337,36 @@ public class DefaultInvoiceDao extends EntityDaoBase<InvoiceModelDao, Invoice, I
                     invoiceByInvoiceId.put(invoiceModelDao.getId(), invoiceModelDao);
                     final boolean isNotShellInvoice = invoiceIdsReferencedFromItems.remove(invoiceModelDao.getId());
 
-                    // Create the invoice if this is not a shell invoice and it does not already exist
-                    if (isNotShellInvoice && invoiceSqlDao.getById(invoiceModelDao.getId().toString(), context) == null) {
-                        createAndRefresh(invoiceSqlDao, invoiceModelDao, context);
-                        createdInvoiceIds.add(invoiceModelDao.getId());
+                    final InvoiceModelDao invoiceOnDisk = invoiceSqlDao.getById(invoiceModelDao.getId().toString(), context);
+                    if (isNotShellInvoice) {
+                        // Create the invoice if this is not a shell invoice and it does not already exist
+                        if (invoiceOnDisk == null) {
+                            createAndRefresh(invoiceSqlDao, invoiceModelDao, context);
+                            createdInvoiceIds.add(invoiceModelDao.getId());
+                        } else if (invoiceOnDisk.getStatus() == InvoiceStatus.DRAFT && invoiceModelDao.getStatus() == InvoiceStatus.COMMITTED) {
+                            invoiceSqlDao.updateStatus(invoiceModelDao.getId().toString(), InvoiceStatus.COMMITTED.toString(), context);
+                            committedReusedInvoiceId.add(invoiceModelDao.getId());
+                        }
                     }
 
                     // Create the invoice items if needed (note: they may not necessarily belong to that invoice)
                     for (final InvoiceItemModelDao invoiceItemModelDao : invoiceModelDao.getInvoiceItems()) {
                         if (transInvoiceItemSqlDao.getById(invoiceItemModelDao.getId().toString(), context) == null) {
                             createdInvoiceItems.add(createInvoiceItemFromTransaction(transInvoiceItemSqlDao, invoiceItemModelDao, context));
-                            modifiedInvoiceIds.add(invoiceItemModelDao.getInvoiceId());
+                            allInvoiceIds.add(invoiceItemModelDao.getInvoiceId());
                         }
                     }
 
-                    final boolean wasInvoiceCreated = createdInvoiceIds.contains(invoiceModelDao.getId());
+                    final boolean wasInvoiceCreatedOrCommitted = createdInvoiceIds.contains(invoiceModelDao.getId()) ||
+                                                       committedReusedInvoiceId.contains(invoiceModelDao.getId());
                     if (InvoiceStatus.COMMITTED.equals(invoiceModelDao.getStatus())) {
-                        committedInvoiceIds.add(invoiceModelDao.getId());
 
-                        if (wasInvoiceCreated) {
+                        if (wasInvoiceCreatedOrCommitted) {
                             notifyBusOfInvoiceCreation(entitySqlDaoWrapperFactory, invoiceModelDao, context);
+                        } else {
+                            adjustedCommittedInvoiceIds.add(invoiceModelDao.getId());
                         }
-                    } else if (wasInvoiceCreated && invoiceModelDao.isParentInvoice()) {
+                    } else if (wasInvoiceCreatedOrCommitted && invoiceModelDao.isParentInvoice()) {
                         // Commit queue
                         notifyOfParentInvoiceCreation(entitySqlDaoWrapperFactory, invoiceModelDao, context);
                     }
@@ -351,7 +375,7 @@ public class DefaultInvoiceDao extends EntityDaoBase<InvoiceModelDao, Invoice, I
                     notifyOfFutureBillingEvents(entitySqlDaoWrapperFactory, invoiceModelDao.getAccountId(), callbackDateTimePerSubscriptions, context);
                 }
 
-                for (final UUID adjustedInvoiceId : modifiedInvoiceIds) {
+                for (final UUID adjustedInvoiceId : allInvoiceIds) {
                     final boolean newInvoice = createdInvoiceIds.contains(adjustedInvoiceId);
                     if (newInvoice) {
                         // New invoice, so no associated payment yet: no need to refresh the invoice state
@@ -362,7 +386,7 @@ public class DefaultInvoiceDao extends EntityDaoBase<InvoiceModelDao, Invoice, I
                         cbaDao.doCBAComplexityFromTransaction(adjustedInvoiceId, invoicesTags, entitySqlDaoWrapperFactory, context);
                     }
 
-                    if (committedInvoiceIds.contains(adjustedInvoiceId) && !newInvoice) {
+                    if (adjustedCommittedInvoiceIds.contains(adjustedInvoiceId)) {
                         // Notify the bus since the balance of the invoice changed (only if the invoice is COMMITTED)
                         notifyBusOfInvoiceAdjustment(entitySqlDaoWrapperFactory, adjustedInvoiceId, accountId, context.getUserToken(), context);
                     }

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/dao/InvoiceDao.java b/invoice/src/main/java/org/killbill/billing/invoice/dao/InvoiceDao.java
index 377b150..3f12d60 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/dao/InvoiceDao.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/dao/InvoiceDao.java
@@ -48,6 +48,8 @@ public interface InvoiceDao extends EntityDao<InvoiceModelDao, Invoice, InvoiceA
     public void setFutureAccountNotificationsForEmptyInvoice(final UUID accountId, final FutureAccountNotifications callbackDateTimePerSubscriptions,
                                                              final InternalCallContext context);
 
+    InvoiceModelDao getEarliestDraftInvoiceByAccount(InternalTenantContext context);
+
     InvoiceModelDao getByNumber(Integer number, InternalTenantContext context) throws InvoiceApiException;
 
     List<InvoiceModelDao> getInvoicesByAccount(InternalTenantContext context);

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/dao/InvoiceSqlDao.java b/invoice/src/main/java/org/killbill/billing/invoice/dao/InvoiceSqlDao.java
index 6d405b0..3acaab5 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/dao/InvoiceSqlDao.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/dao/InvoiceSqlDao.java
@@ -59,5 +59,8 @@ public interface InvoiceSqlDao extends EntitySqlDao<InvoiceModelDao, Invoice> {
     @SqlQuery
     List<InvoiceModelDao> getByIds(@BindIn("ids") final Collection<String> invoiceIds,
                                    @SmartBindBean final InternalTenantContext context);
+    @SqlQuery
+    InvoiceModelDao getEarliestDraftInvoiceByAccount(@SmartBindBean final InternalTenantContext context);
+
 }
 

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 1d73b89..69bb538 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
@@ -66,10 +66,13 @@ public class DefaultInvoiceGenerator implements InvoiceGenerator {
      * adjusts target date to the maximum invoice target date, if future invoices exist
      */
     @Override
-    public InvoiceWithMetadata generateInvoice(final ImmutableAccountData account, @Nullable final BillingEventSet events,
+    public InvoiceWithMetadata generateInvoice(final ImmutableAccountData account,
+                                               @Nullable final BillingEventSet events,
                                                @Nullable final List<Invoice> existingInvoices,
+                                               @Nullable final UUID targetInvoiceId,
                                                final LocalDate targetDate,
-                                               final Currency targetCurrency, final InternalCallContext context) throws InvoiceApiException {
+                                               final Currency targetCurrency,
+                                               final InternalCallContext context) throws InvoiceApiException {
         if ((events == null) || (events.size() == 0) || events.isAccountAutoInvoiceOff()) {
             return new InvoiceWithMetadata(null, ImmutableMap.<UUID, SubscriptionFutureNotificationDates>of());
         }
@@ -79,14 +82,16 @@ public class DefaultInvoiceGenerator implements InvoiceGenerator {
 
         final LocalDate invoiceDate = context.toLocalDate(context.getCreatedDate());
         final InvoiceStatus invoiceStatus = events.isAccountAutoInvoiceDraft() ? InvoiceStatus.DRAFT : InvoiceStatus.COMMITTED;
-        final DefaultInvoice invoice = new DefaultInvoice(account.getId(), invoiceDate, adjustedTargetDate, targetCurrency, invoiceStatus);
-        final UUID invoiceId = invoice.getId();
+        final DefaultInvoice invoice = targetInvoiceId != null ?
+                                       new DefaultInvoice(targetInvoiceId, account.getId(), null, invoiceDate, adjustedTargetDate, targetCurrency, false, invoiceStatus) :
+                                       new DefaultInvoice(account.getId(), invoiceDate, adjustedTargetDate, targetCurrency, invoiceStatus);
+
         final Map<UUID, SubscriptionFutureNotificationDates> perSubscriptionFutureNotificationDates = new HashMap<UUID, SubscriptionFutureNotificationDates>();
 
-        final List<InvoiceItem> fixedAndRecurringItems = recurringInvoiceItemGenerator.generateItems(account, invoiceId, events, existingInvoices, adjustedTargetDate, targetCurrency, perSubscriptionFutureNotificationDates, context);
+        final List<InvoiceItem> fixedAndRecurringItems = recurringInvoiceItemGenerator.generateItems(account, invoice.getId(), events, existingInvoices, adjustedTargetDate, targetCurrency, perSubscriptionFutureNotificationDates, context);
         invoice.addInvoiceItems(fixedAndRecurringItems);
 
-        final List<InvoiceItem> usageItems = usageInvoiceItemGenerator.generateItems(account, invoiceId, events, existingInvoices, adjustedTargetDate, targetCurrency, perSubscriptionFutureNotificationDates, context);
+        final List<InvoiceItem> usageItems = usageInvoiceItemGenerator.generateItems(account, invoice.getId(), events, existingInvoices, adjustedTargetDate, targetCurrency, perSubscriptionFutureNotificationDates, context);
         invoice.addInvoiceItems(usageItems);
 
 

diff --git a/invoice/src/main/java/org/killbill/billing/invoice/generator/InvoiceGenerator.java b/invoice/src/main/java/org/killbill/billing/invoice/generator/InvoiceGenerator.java
index e513a00..fac2799 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/generator/InvoiceGenerator.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/generator/InvoiceGenerator.java
@@ -17,6 +17,7 @@
 package org.killbill.billing.invoice.generator;
 
 import java.util.List;
+import java.util.UUID;
 
 import javax.annotation.Nullable;
 
@@ -30,6 +31,6 @@ import org.killbill.billing.invoice.api.InvoiceApiException;
 import org.killbill.billing.junction.BillingEventSet;
 
 public interface InvoiceGenerator {
-    InvoiceWithMetadata generateInvoice(ImmutableAccountData account,  @Nullable BillingEventSet events, @Nullable List<Invoice> existingInvoices,
-                                        LocalDate targetDate, Currency targetCurrency, final InternalCallContext context) throws InvoiceApiException;
+    InvoiceWithMetadata generateInvoice(ImmutableAccountData account, @Nullable BillingEventSet events, @Nullable List<Invoice> existingInvoices,
+                                        final UUID targetInvoiceId, LocalDate targetDate, Currency targetCurrency, final InternalCallContext context) throws InvoiceApiException;
 }

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 d8fc889..4f274ba 100644
--- a/invoice/src/main/java/org/killbill/billing/invoice/InvoiceDispatcher.java
+++ b/invoice/src/main/java/org/killbill/billing/invoice/InvoiceDispatcher.java
@@ -382,7 +382,16 @@ public class InvoiceDispatcher {
                                                                                                 }));
 
             final Currency targetCurrency = account.getCurrency();
-            final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, billingEvents, invoices, targetDate, targetCurrency, context);
+
+            final UUID targetInvoiceId;
+            if (billingEvents.isAccountAutoInvoiceReuseDraft()) {
+                final InvoiceModelDao earliestDraftInvoice = invoiceDao.getEarliestDraftInvoiceByAccount(context);
+                targetInvoiceId = earliestDraftInvoice != null  ? earliestDraftInvoice.getId() : null;
+            } else {
+                targetInvoiceId = null;
+            }
+
+            final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, billingEvents, invoices, targetInvoiceId, targetDate, targetCurrency, context);
             final DefaultInvoice invoice = invoiceWithMetadata.getInvoice();
 
             // Compute future notifications

diff --git a/invoice/src/main/resources/org/killbill/billing/invoice/dao/InvoiceSqlDao.sql.stg b/invoice/src/main/resources/org/killbill/billing/invoice/dao/InvoiceSqlDao.sql.stg
index ab962d9..141dca0 100644
--- a/invoice/src/main/resources/org/killbill/billing/invoice/dao/InvoiceSqlDao.sql.stg
+++ b/invoice/src/main/resources/org/killbill/billing/invoice/dao/InvoiceSqlDao.sql.stg
@@ -71,6 +71,17 @@ getParentDraftInvoice() ::= <<
    <defaultOrderBy("")>
 >>
 
+getEarliestDraftInvoiceByAccount() ::= <<
+  SELECT <allTableFields("")>
+    FROM <tableName()>
+   WHERE <accountRecordIdField("")> = :accountRecordId
+     AND status = 'DRAFT'
+   <AND_CHECK_TENANT("")>
+   <defaultOrderBy("")>
+   limit 1
+;
+>>
+
 getByIds(ids) ::= <<
 select
   <allTableFields("t.")>

diff --git a/invoice/src/test/java/org/killbill/billing/invoice/dao/MockInvoiceDao.java b/invoice/src/test/java/org/killbill/billing/invoice/dao/MockInvoiceDao.java
index 0e1ea27..3cf8c32 100644
--- a/invoice/src/test/java/org/killbill/billing/invoice/dao/MockInvoiceDao.java
+++ b/invoice/src/test/java/org/killbill/billing/invoice/dao/MockInvoiceDao.java
@@ -81,6 +81,11 @@ public class MockInvoiceDao extends MockEntityDaoBase<InvoiceModelDao, Invoice, 
     }
 
     @Override
+    public InvoiceModelDao getEarliestDraftInvoiceByAccount(final InternalTenantContext context) {
+        return null;
+    }
+
+    @Override
     public List<InvoiceItemModelDao> createInvoices(final List<InvoiceModelDao> invoiceModelDaos, final InternalCallContext context) {
         synchronized (monitor) {
             final List<InvoiceItemModelDao> createdItems = new LinkedList<InvoiceItemModelDao>();

diff --git a/invoice/src/test/java/org/killbill/billing/invoice/dao/TestInvoiceDao.java b/invoice/src/test/java/org/killbill/billing/invoice/dao/TestInvoiceDao.java
index 7199e5d..ec684f1 100644
--- a/invoice/src/test/java/org/killbill/billing/invoice/dao/TestInvoiceDao.java
+++ b/invoice/src/test/java/org/killbill/billing/invoice/dao/TestInvoiceDao.java
@@ -502,6 +502,42 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
     }
 
     @Test(groups = "slow")
+    public void testGetEarliestDraftInvoiceByAccount() throws EntityPersistenceException {
+        final UUID accountId = account.getId();
+
+        InvoiceModelDao result;
+        result = invoiceDao.getEarliestDraftInvoiceByAccount(context);
+        assertNull(result);
+
+        final LocalDate targetDate1 = new LocalDate(2011, 10, 6);
+        final Invoice invoice1 = new DefaultInvoice(accountId, clock.getUTCToday(), targetDate1, Currency.USD, InvoiceStatus.DRAFT);
+        invoiceUtil.createInvoice(invoice1, context);
+
+        result = invoiceDao.getEarliestDraftInvoiceByAccount(context);
+        assertNotNull(result);
+        assertEquals(result.getId(), invoice1.getId());
+
+        final LocalDate targetDate2 = new LocalDate(2011, 12, 6);
+        final Invoice invoice2 = new DefaultInvoice(accountId, clock.getUTCToday(), targetDate2, Currency.USD);
+        invoiceUtil.createInvoice(invoice2, context);
+
+        result = invoiceDao.getEarliestDraftInvoiceByAccount(context);
+        assertNotNull(result);
+        assertEquals(result.getId(), invoice1.getId());
+
+
+        final LocalDate targetDate3 = new LocalDate(2011, 10, 6);
+        final Invoice invoice3 = new DefaultInvoice(accountId, clock.getUTCToday(), targetDate3, Currency.USD, InvoiceStatus.DRAFT);
+        invoiceUtil.createInvoice(invoice3, context);
+
+        result = invoiceDao.getEarliestDraftInvoiceByAccount(context);
+        assertNotNull(result);
+        assertEquals(result.getId(), invoice1.getId());
+    }
+
+
+
+    @Test(groups = "slow")
     public void testAccountBalance() throws EntityPersistenceException {
         final UUID accountId = account.getId();
         final UUID bundleId = UUID.randomUUID();
@@ -1207,7 +1243,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
         final BillingEventSet events = new MockBillingEventSet();
         events.add(event1);
 
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, invoiceList, targetDate, Currency.USD, context);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, invoiceList, null, targetDate, Currency.USD, context);
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();
         assertEquals(invoice1.getBalance(), KillBillMoney.of(TEN, invoice1.getCurrency()));
         invoiceList.add(invoice1);
@@ -1226,7 +1262,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
 
         // second invoice should be for one half (14/28 days) the difference between the rate plans
         // this is a temporary state, since it actually contains an adjusting item that properly belong to invoice 1
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, targetDate, Currency.USD, context);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, null, targetDate, Currency.USD, context);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
 
         assertEquals(invoice2.getBalance(), KillBillMoney.of(FIVE, invoice2.getCurrency()));
@@ -1260,7 +1296,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
         events.add(event);
 
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 1, 15);
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, context);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, context);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
 
@@ -1302,7 +1338,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
         events.add(event1);
 
         final UUID accountId = account.getId();
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, new LocalDate(effectiveDate1), Currency.USD, context);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, null, new LocalDate(effectiveDate1), Currency.USD, context);
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();
         assertNotNull(invoice1);
         assertEquals(invoice1.getNumberOfItems(), 1);
@@ -1319,7 +1355,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
                                                                        "testEvent2", 2L, SubscriptionBaseTransitionType.PHASE);
         events.add(event2);
 
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, new LocalDate(effectiveDate2), Currency.USD, context);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, null, new LocalDate(effectiveDate2), Currency.USD, context);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
         assertNotNull(invoice2);
         assertEquals(invoice2.getNumberOfItems(), 1);
@@ -1330,7 +1366,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
         //invoiceUtil.createInvoice(invoice2, invoice2.getTargetDate().getDayOfMonth(), callcontext);
 
         final DateTime effectiveDate3 = effectiveDate2.plusMonths(1);
-        final InvoiceWithMetadata invoiceWithMetadata3 = generator.generateInvoice(account, events, invoiceList, new LocalDate(effectiveDate3), Currency.USD, context);
+        final InvoiceWithMetadata invoiceWithMetadata3 = generator.generateInvoice(account, events, invoiceList, null, new LocalDate(effectiveDate3), Currency.USD, context);
         final Invoice invoice3 = invoiceWithMetadata3.getInvoice();
         assertNotNull(invoice3);
         assertEquals(invoice3.getNumberOfItems(), 1);
@@ -1342,7 +1378,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
     @Test(groups = "slow")
     public void testInvoiceForEmptyEventSet() throws InvoiceApiException {
         final BillingEventSet events = new MockBillingEventSet();
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, new LocalDate(), Currency.USD, context);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, new LocalDate(), Currency.USD, context);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNull(invoice);
     }
@@ -1377,7 +1413,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
                                                                        "testEvent2", 2L, SubscriptionBaseTransitionType.CHANGE);
         events.add(event2);
 
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, new LocalDate(effectiveDate2), Currency.USD, context);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, new LocalDate(effectiveDate2), Currency.USD, context);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), 2);
@@ -1449,7 +1485,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
                                                                        BillingPeriod.MONTHLY, 31, BillingMode.IN_ADVANCE,
                                                                        "new-event", 1L, SubscriptionBaseTransitionType.CREATE);
         events.add(event1);
-        final InvoiceWithMetadata newInvoiceWithMetadata = generator.generateInvoice(account, events, invoices, targetDate, Currency.USD, context);
+        final InvoiceWithMetadata newInvoiceWithMetadata = generator.generateInvoice(account, events, invoices, null, targetDate, Currency.USD, context);
         final Invoice newInvoice = newInvoiceWithMetadata.getInvoice();
         invoiceUtil.createInvoice(newInvoice, context);
 
@@ -1485,7 +1521,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
                                                                        "testEvent1", 1L, SubscriptionBaseTransitionType.CHANGE);
         events.add(event1);
 
-        InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, invoices, new LocalDate(targetDate1), Currency.USD, context);
+        InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, invoices, null, new LocalDate(targetDate1), Currency.USD, context);
         Invoice invoice1 = invoiceWithMetadata1.getInvoice();
         invoices.add(invoice1);
         invoiceUtil.createInvoice(invoice1, context);
@@ -1497,7 +1533,7 @@ public class TestInvoiceDao extends InvoiceTestSuiteWithEmbeddedDB {
                                                                        BillingPeriod.MONTHLY, 31, BillingMode.IN_ADVANCE,
                                                                        "testEvent2", 2L, SubscriptionBaseTransitionType.CHANGE);
         events.add(event2);
-        InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoices, new LocalDate(targetDate2), Currency.USD, context);
+        InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoices, null, new LocalDate(targetDate2), Currency.USD, context);
         Invoice invoice2 = invoiceWithMetadata2.getInvoice();
         invoiceUtil.createInvoice(invoice2, context);
         invoice2 = new DefaultInvoice(invoiceDao.getById(invoice2.getId(), context));

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 bfc6eb7..d7aa3d4 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
@@ -127,14 +127,14 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
     @Test(groups = "fast")
     public void testWithNullEventSetAndNullInvoiceSet() throws InvoiceApiException {
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, null, null, clock.getUTCToday(), Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, null, null, null, clock.getUTCToday(), Currency.USD, internalCallContext);
         assertNull(invoiceWithMetadata.getInvoice());
     }
 
     @Test(groups = "fast")
     public void testWithEmptyEventSet() throws InvoiceApiException {
         final BillingEventSet events = new MockBillingEventSet();
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, clock.getUTCToday(), Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, clock.getUTCToday(), Currency.USD, internalCallContext);
         assertNull(invoiceWithMetadata.getInvoice());
     }
 
@@ -153,7 +153,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event);
 
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 10, 3);
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), 2);
@@ -184,7 +184,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event);
 
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 10, 3);
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), 2);
@@ -217,7 +217,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         // Target date is the next BCD, in local time
         final LocalDate targetDate = invoiceUtil.buildDate(2012, 8, bcdLocal);
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), 2);
@@ -241,7 +241,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         // Set a target date of today (start date)
         final LocalDate targetDate = startDate;
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), 1);
@@ -263,7 +263,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event);
 
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 10, 3);
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), 2);
@@ -295,7 +295,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event2);
 
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 10, 3);
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), 2);
@@ -321,7 +321,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 12, 3);
         final UUID accountId = UUID.randomUUID();
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), 4);
@@ -362,7 +362,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event3);
 
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 12, 3);
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), 4);
@@ -384,13 +384,13 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event1);
 
         LocalDate targetDate = invoiceUtil.buildDate(2011, 12, 1);
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final List<Invoice> existingInvoices = new ArrayList<Invoice>();
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();
         existingInvoices.add(invoice1);
 
         targetDate = invoiceUtil.buildDate(2011, 12, 3);
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, existingInvoices, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, existingInvoices, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
         assertNull(invoice2);
     }
@@ -563,7 +563,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 1, 1);
         events.add(createBillingEvent(UUID.randomUUID(), UUID.randomUUID(), targetDate, plan, planPhase, 1));
 
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getInvoiceItems().size(), 1);
@@ -580,7 +580,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         events.add(createBillingEvent(UUID.randomUUID(), UUID.randomUUID(), startDate, plan, planPhase, startDate.getDayOfMonth()));
 
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         final RecurringInvoiceItem item = (RecurringInvoiceItem) invoice.getInvoiceItems().get(0);
 
@@ -618,14 +618,14 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         events.add(event2);
         events.add(event1);
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, new LocalDate("2012-02-01"), Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, null, new LocalDate("2012-02-01"), Currency.USD, internalCallContext);
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();
         assertNotNull(invoice1);
         assertEquals(invoice1.getNumberOfItems(), 1);
 
         final List<Invoice> invoiceList = new ArrayList<Invoice>();
         invoiceList.add(invoice1);
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, new LocalDate("2012-04-05"), Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, null, new LocalDate("2012-04-05"), Currency.USD, internalCallContext);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
         assertNotNull(invoice2);
         assertEquals(invoice2.getNumberOfItems(), 1);
@@ -650,7 +650,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event1);
 
         // ensure both components are invoiced
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, startDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, null, startDate, Currency.USD, internalCallContext);
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();
         assertNotNull(invoice1);
         assertEquals(invoice1.getNumberOfItems(), 2);
@@ -663,7 +663,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         final LocalDate currentDate = startDate.plusMonths(1);
 
         // ensure that only the recurring price is invoiced
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, currentDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, null, currentDate, Currency.USD, internalCallContext);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
         assertNotNull(invoice2);
         assertEquals(invoice2.getNumberOfItems(), 1);
@@ -689,7 +689,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event1);
 
         // ensure that a single invoice item is generated for the fixed cost
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, startDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, null, startDate, Currency.USD, internalCallContext);
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();assertNotNull(invoice1);
         assertEquals(invoice1.getNumberOfItems(), 1);
         assertEquals(invoice1.getBalance(), KillBillMoney.of(fixedCost1, invoice1.getCurrency()));
@@ -703,7 +703,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event2);
 
         // ensure that a single invoice item is generated for the fixed cost
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, phaseChangeDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, null, phaseChangeDate, Currency.USD, internalCallContext);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
         assertEquals(invoice2.getNumberOfItems(), 1);
         assertEquals(invoice2.getBalance(), KillBillMoney.of(fixedCost2, invoice2.getCurrency()));
@@ -735,7 +735,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         final LocalDate discountPhaseEndDate = trialPhaseEndDate.plusMonths(6);
         events.add(createBillingEvent(subscriptionId, bundleId, discountPhaseEndDate, plan1, phase3, BILL_CYCLE_DAY));
 
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, creationDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, null, creationDate, Currency.USD, internalCallContext);
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();
         assertNotNull(invoice1);
         assertEquals(invoice1.getNumberOfItems(), 1);
@@ -744,7 +744,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         final List<Invoice> invoiceList = new ArrayList<Invoice>();
         invoiceList.add(invoice1);
 
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, trialPhaseEndDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoiceList, null, trialPhaseEndDate, Currency.USD, internalCallContext);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
         assertNotNull(invoice2);
         assertEquals(invoice2.getNumberOfItems(), 1);
@@ -753,7 +753,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         invoiceList.add(invoice2);
         LocalDate targetDate = new LocalDate(trialPhaseEndDate.getYear(), trialPhaseEndDate.getMonthOfYear(), BILL_CYCLE_DAY);
-        final InvoiceWithMetadata invoiceWithMetadata3 = generator.generateInvoice(account, events, invoiceList, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata3 = generator.generateInvoice(account, events, invoiceList, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice3 = invoiceWithMetadata3.getInvoice();
         assertNotNull(invoice3);
         assertEquals(invoice3.getNumberOfItems(), 1);
@@ -762,7 +762,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         invoiceList.add(invoice3);
         targetDate = targetDate.plusMonths(6);
-        final InvoiceWithMetadata invoiceWithMetadata4 = generator.generateInvoice(account, events, invoiceList, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata4 = generator.generateInvoice(account, events, invoiceList, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice4 = invoiceWithMetadata4.getInvoice();
         assertNotNull(invoice4);
         assertEquals(invoice4.getNumberOfItems(), 7);
@@ -775,7 +775,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         final Plan plan1 = new MockPlan();
         final PlanPhase phase1 = createMockMonthlyPlanPhase(null, ZERO, PhaseType.TRIAL);
         events.add(createBillingEvent(UUID.randomUUID(), UUID.randomUUID(), clock.getUTCToday(), plan1, phase1, 1));
-        generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
     }
 
     @Test(groups = "fast")
@@ -795,7 +795,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(createBillingEvent(baseSubscription.getId(), baseSubscription.getBundleId(), april25, basePlan, basePlanEvergreen, 25));
 
         // generate invoice
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, april25, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, null, april25, Currency.USD, internalCallContext);
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();
         assertNotNull(invoice1);
         assertEquals(invoice1.getNumberOfItems(), 1);
@@ -817,7 +817,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(createBillingEvent(addOnSubscription2.getId(), baseSubscription.getBundleId(), april28, addOn2Plan, addOn2PlanPhaseEvergreen, 25));
 
         // generate invoice
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoices, april28, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoices, null, april28, Currency.USD, internalCallContext);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
 
         invoices.add(invoice2);
@@ -836,7 +836,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         // generate invoice
         final LocalDate may1 = new LocalDate(2012, 5, 1);
-        final InvoiceWithMetadata invoiceWithMetadata3 = generator.generateInvoice(account, newEvents, invoices, may1, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata3 = generator.generateInvoice(account, newEvents, invoices, null, may1, Currency.USD, internalCallContext);
         final Invoice invoice3 = invoiceWithMetadata3.getInvoice();
         assertNotNull(invoice3);
         assertEquals(invoice3.getNumberOfItems(), 3);
@@ -859,7 +859,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         final BillingEventSet events = new MockBillingEventSet();
         events.add(createBillingEvent(originalSubscription.getId(), originalSubscription.getBundleId(), april25, originalPlan, originalPlanEvergreen, 25));
 
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, april25, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, events, null, null, april25, Currency.USD, internalCallContext);
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();
 
         printDetailInvoice(invoice1);
@@ -882,7 +882,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(createBillingEvent(newSubscription.getId(), originalSubscription.getBundleId(), april25, newPlan, newPlanEvergreen, 25));
 
         // generate a new invoice
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoices, april25, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, invoices, null, april25, Currency.USD, internalCallContext);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
 
         printDetailInvoice(invoice2);
@@ -947,7 +947,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         // Generate a new invoice
 
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, existingInvoices, targetDate, currency, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, existingInvoices, null, targetDate, currency, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertEquals(invoice.getNumberOfItems(), 7);
         assertEquals(invoice.getInvoiceItems().get(0).getInvoiceItemType(), InvoiceItemType.RECURRING);
@@ -982,7 +982,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         existingInvoices.add(invoice);
 
         // Generate next invoice (no-op)
-        final InvoiceWithMetadata newInvoiceWithMetdata = generator.generateInvoice(account, events, existingInvoices, targetDate, currency, internalCallContext);
+        final InvoiceWithMetadata newInvoiceWithMetdata = generator.generateInvoice(account, events, existingInvoices, null, targetDate, currency, internalCallContext);
         final Invoice newInvoice = newInvoiceWithMetdata.getInvoice();
         assertNull(newInvoice);
     }
@@ -1004,7 +1004,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 10, 3);
         final UUID accountId = UUID.randomUUID();
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
 
         assertNull(invoiceWithMetadata.getInvoice());
     }
@@ -1033,7 +1033,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         eventSet.add(createBillingEvent(subscriptionId2, bundleId, startDate, plan2, plan2phase1, 1));
 
         // generate the first invoice
-        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, eventSet, invoices, startDate, currency, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata1 = generator.generateInvoice(account, eventSet, invoices, null, startDate, currency, internalCallContext);
         final Invoice invoice1 = invoiceWithMetadata1.getInvoice();
         assertNotNull(invoice1);
         assertTrue(invoice1.getBalance().compareTo(FIFTEEN.add(TWELVE)) == 0);
@@ -1045,7 +1045,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         eventSet.addSubscriptionWithAutoInvoiceOff(subscriptionId1);
 
         final LocalDate targetDate2 = startDate.plusMonths(1);
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, eventSet, invoices, targetDate2, currency, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, eventSet, invoices, null, targetDate2, currency, internalCallContext);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
         assertNotNull(invoice2);
         assertTrue(invoice2.getBalance().compareTo(TWELVE) == 0);
@@ -1054,7 +1054,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         final LocalDate targetDate3 = targetDate2.plusMonths(1);
         eventSet.clearSubscriptionsWithAutoInvoiceOff();
         eventSet.add(subscription1creation);
-        final InvoiceWithMetadata invoiceWithMetadata3 = generator.generateInvoice(account, eventSet, invoices, targetDate3, currency, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata3 = generator.generateInvoice(account, eventSet, invoices, null, targetDate3, currency, internalCallContext);
         final Invoice invoice3 = invoiceWithMetadata3.getInvoice();
         assertNotNull(invoice3);
         assertTrue(invoice3.getBalance().compareTo(FIFTEEN.multiply(TWO).add(TWELVE)) == 0);
@@ -1077,7 +1077,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         final LocalDate targetDate = invoiceUtil.buildDate(2011, 10, 3);
         final UUID accountId = UUID.randomUUID();
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
 
         assertNotNull(invoiceWithMetadata.getInvoice());
         assertEquals(invoiceWithMetadata.getInvoice().getStatus(), InvoiceStatus.DRAFT);
@@ -1104,7 +1104,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
         events.add(event);
 
         final LocalDate targetDate = invoiceUtil.buildDate(2016, 10, 9);
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, null, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
 
         assertNotNull(invoice);
@@ -1130,7 +1130,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
         final List<Invoice> existingInvoices = new ArrayList<Invoice>();
         existingInvoices.add(invoice);
-        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, existingInvoices, targetDate, Currency.USD, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata2 = generator.generateInvoice(account, events, existingInvoices, null, targetDate, Currency.USD, internalCallContext);
         final Invoice invoice2 = invoiceWithMetadata2.getInvoice();
 
         assertNull(invoice2);
@@ -1392,7 +1392,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
                                        final LocalDate targetDate, final int expectedNumberOfItems,
                                        final BigDecimal expectedAmount) throws InvoiceApiException {
         final Currency currency = Currency.USD;
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, existingInvoices, targetDate, currency, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, existingInvoices, null, targetDate, currency, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNotNull(invoice);
         assertEquals(invoice.getNumberOfItems(), expectedNumberOfItems);
@@ -1404,7 +1404,7 @@ public class TestDefaultInvoiceGenerator extends InvoiceTestSuiteNoDB {
 
     private void testNullInvoiceGeneration(final BillingEventSet events, final List<Invoice> existingInvoices, final LocalDate targetDate) throws InvoiceApiException {
         final Currency currency = Currency.USD;
-        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, existingInvoices, targetDate, currency, internalCallContext);
+        final InvoiceWithMetadata invoiceWithMetadata = generator.generateInvoice(account, events, existingInvoices, null, targetDate, currency, internalCallContext);
         final Invoice invoice = invoiceWithMetadata.getInvoice();
         assertNull(invoice);
     }

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 dabf129..5dac9bb 100644
--- a/invoice/src/test/java/org/killbill/billing/invoice/MockBillingEventSet.java
+++ b/invoice/src/test/java/org/killbill/billing/invoice/MockBillingEventSet.java
@@ -36,6 +36,7 @@ public class MockBillingEventSet extends TreeSet<BillingEvent> implements Billin
 
     private boolean isAccountInvoiceOff;
     private boolean isAccountAutoInvoiceDraft;
+    private boolean isAccountAutoInvoiceReuseDraft;
 
     private List<UUID> subscriptionIdsWithAutoInvoiceOff;
 
@@ -43,6 +44,7 @@ public class MockBillingEventSet extends TreeSet<BillingEvent> implements Billin
         super();
         this.isAccountInvoiceOff = false;
         this.isAccountAutoInvoiceDraft = false;
+        this.isAccountAutoInvoiceReuseDraft = false;
         this.subscriptionIdsWithAutoInvoiceOff = new ArrayList<UUID>();
     }
 
@@ -61,6 +63,11 @@ public class MockBillingEventSet extends TreeSet<BillingEvent> implements Billin
     }
 
     @Override
+    public boolean isAccountAutoInvoiceReuseDraft() {
+        return isAccountAutoInvoiceReuseDraft;
+    }
+
+    @Override
     public BillingMode getRecurringBillingMode() {
         return BillingMode.IN_ADVANCE;
     }

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 35bda36..7eec199 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
@@ -42,12 +42,14 @@ public class DefaultBillingEventSet extends TreeSet<BillingEvent> implements Sor
 
     private final boolean accountAutoInvoiceOff;
     private final boolean accountAutoInvoiceDraft;
+    private final boolean accountAutoInvoiceReuseDraft;
     private final List<UUID> subscriptionIdsWithAutoInvoiceOff;
     private final BillingMode recurringBillingMode;
 
-    public DefaultBillingEventSet(final boolean accountAutoInvoiceOff, final boolean accountAutoInvoiceDraft, final BillingMode recurringBillingMode) {
+    public DefaultBillingEventSet(final boolean accountAutoInvoiceOff, final boolean accountAutoInvoiceDraft, final boolean accountAutoInvoiceReuseDraft, final BillingMode recurringBillingMode) {
         this.accountAutoInvoiceOff = accountAutoInvoiceOff;
         this.accountAutoInvoiceDraft = accountAutoInvoiceDraft;
+        this.accountAutoInvoiceReuseDraft = accountAutoInvoiceReuseDraft;
         this.recurringBillingMode = recurringBillingMode;
         this.subscriptionIdsWithAutoInvoiceOff = new ArrayList<UUID>();
     }
@@ -63,6 +65,11 @@ public class DefaultBillingEventSet extends TreeSet<BillingEvent> implements Sor
     }
 
     @Override
+    public boolean isAccountAutoInvoiceReuseDraft() {
+        return accountAutoInvoiceReuseDraft;
+    }
+
+    @Override
     public BillingMode getRecurringBillingMode() {
         return recurringBillingMode;
     }

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 b8f0bac..530cf8d 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
@@ -96,17 +96,18 @@ public class DefaultInternalBillingApi implements BillingInternalApi {
         final List<Tag> accountTags = tagApi.getTags(accountId, ObjectType.ACCOUNT, context);
         final boolean found_AUTO_INVOICING_OFF = is_AUTO_INVOICING_OFF(accountTags);
         final boolean found_INVOICING_DRAFT = is_AUTO_INVOICING_DRAFT(accountTags);
+        final boolean found_INVOICING_REUSE_DRAFT = is_AUTO_INVOICING_REUSE_DRAFT(accountTags);
 
         final Set<UUID> skippedSubscriptions = new HashSet<UUID>();
         final DefaultBillingEventSet result;
 
         if (found_AUTO_INVOICING_OFF) {
-            result = new DefaultBillingEventSet(true, found_INVOICING_DRAFT, ((StaticCatalog) currentCatalog).getRecurringBillingMode()); // billing is off, we are done
+            result = new DefaultBillingEventSet(true, found_INVOICING_DRAFT, found_INVOICING_REUSE_DRAFT, ((StaticCatalog) currentCatalog).getRecurringBillingMode()); // billing is off, we are done
         } else {
             final List<SubscriptionBaseBundle> bundles = subscriptionApi.getBundlesForAccount(accountId, context);
 
             final ImmutableAccountData account = accountApi.getImmutableAccountDataById(accountId, context);
-            result = new DefaultBillingEventSet(false, found_INVOICING_DRAFT, ((StaticCatalog) currentCatalog).getRecurringBillingMode());
+            result = new DefaultBillingEventSet(false, found_INVOICING_DRAFT, found_INVOICING_REUSE_DRAFT, ((StaticCatalog) currentCatalog).getRecurringBillingMode());
             addBillingEventsForBundles(bundles, account, dryRunArguments, context, result, skippedSubscriptions, currentCatalog);
         }
 
@@ -259,4 +260,15 @@ public class DefaultInternalBillingApi implements BillingInternalApi {
             }
         });
     }
+
+    private boolean is_AUTO_INVOICING_REUSE_DRAFT(final List<Tag> tags) {
+        return Iterables.any(tags, new Predicate<Tag>() {
+            @Override
+            public boolean apply(final Tag input) {
+                return input.getTagDefinitionId().equals(ControlTagType.AUTO_INVOICING_REUSE_DRAFT.getId());
+            }
+        });
+    }
+
+
 }