killbill-uncached

diff --git a/invoice/src/main/java/com/ning/billing/invoice/api/svcs/DefaultInvoiceInternalApi.java b/invoice/src/main/java/com/ning/billing/invoice/api/svcs/DefaultInvoiceInternalApi.java
index 4148e00..7294f86 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/api/svcs/DefaultInvoiceInternalApi.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/api/svcs/DefaultInvoiceInternalApi.java
@@ -49,6 +49,7 @@ import com.google.common.collect.Collections2;
 public class DefaultInvoiceInternalApi implements InvoiceInternalApi {
 
     private static final WithInvoiceApiException<InvoicePayment> invoicePaymentWithException = new WithInvoiceApiException<InvoicePayment>();
+    private static final WithInvoiceApiException<Void> voidWithException = new WithInvoiceApiException<Void>();
 
     private final InvoiceDao dao;
 
@@ -132,19 +133,22 @@ public class DefaultInvoiceInternalApi implements InvoiceInternalApi {
                 if (amount.compareTo(BigDecimal.ZERO) <= 0) {
                     throw new InvoiceApiException(ErrorCode.PAYMENT_REFUND_AMOUNT_NEGATIVE_OR_NULL);
                 }
-
-                final Collection<InvoicePayment> invoicePayments = Collections2.transform(dao.getInvoicePayments(paymentId, context), new Function<InvoicePaymentModelDao, InvoicePayment>() {
-                    @Override
-                    public InvoicePayment apply(final InvoicePaymentModelDao input) {
-                        return new DefaultInvoicePayment(input);
-                    }
-                });
-
                 return new DefaultInvoicePayment(dao.createRefund(paymentId, amount, isInvoiceAdjusted, invoiceItemIdsWithAmounts, paymentCookieId, context));
             }
         });
     }
 
+    @Override
+    public void consumeExistingCBAOnAccountWithUnpaidInvoices(final UUID accountId, final InternalCallContext context) throws InvoiceApiException {
+        voidWithException.executeAndThrow(new WithInvoiceApiExceptionCallback<Void>()  {
+            @Override
+            public Void doHandle() throws InvoiceApiException {
+                dao.consumeExstingCBAOnAccountWithUnpaidInvoices(accountId, context);
+                return null;
+            }
+        });
+    }
+
     //
     // Allow to safely catch TransactionFailedException exceptions and rethrow the correct InvoiceApiException exception
     //

diff --git a/payment/src/main/java/com/ning/billing/payment/core/PaymentMethodProcessor.java b/payment/src/main/java/com/ning/billing/payment/core/PaymentMethodProcessor.java
index 0992e64..3bd3665 100644
--- a/payment/src/main/java/com/ning/billing/payment/core/PaymentMethodProcessor.java
+++ b/payment/src/main/java/com/ning/billing/payment/core/PaymentMethodProcessor.java
@@ -49,6 +49,7 @@ import com.ning.billing.util.callcontext.InternalCallContext;
 import com.ning.billing.util.callcontext.InternalTenantContext;
 import com.ning.billing.util.globallocker.GlobalLocker;
 import com.ning.billing.util.svcapi.account.AccountInternalApi;
+import com.ning.billing.util.svcapi.invoice.InvoiceInternalApi;
 import com.ning.billing.util.svcapi.tag.TagInternalApi;
 import com.ning.billing.util.svcsapi.bus.InternalBus;
 
@@ -67,12 +68,13 @@ public class PaymentMethodProcessor extends ProcessorBase {
     @Inject
     public PaymentMethodProcessor(final OSGIServiceRegistration<PaymentPluginApi> pluginRegistry,
                                   final AccountInternalApi accountInternalApi,
+                                  final InvoiceInternalApi invoiceApi,
                                   final InternalBus eventBus,
                                   final PaymentDao paymentDao,
                                   final TagInternalApi tagUserApi,
                                   final GlobalLocker locker,
                                   @Named(PLUGIN_EXECUTOR_NAMED) final ExecutorService executor) {
-        super(pluginRegistry, accountInternalApi, eventBus, paymentDao, tagUserApi, locker, executor);
+        super(pluginRegistry, accountInternalApi, eventBus, paymentDao, tagUserApi, locker, executor, invoiceApi);
     }
 
     public Set<String> getAvailablePlugins() {

diff --git a/payment/src/main/java/com/ning/billing/payment/core/PaymentProcessor.java b/payment/src/main/java/com/ning/billing/payment/core/PaymentProcessor.java
index b221108..016c534 100644
--- a/payment/src/main/java/com/ning/billing/payment/core/PaymentProcessor.java
+++ b/payment/src/main/java/com/ning/billing/payment/core/PaymentProcessor.java
@@ -78,7 +78,6 @@ import static com.ning.billing.payment.glue.PaymentModule.PLUGIN_EXECUTOR_NAMED;
 public class PaymentProcessor extends ProcessorBase {
 
     private final PaymentMethodProcessor paymentMethodProcessor;
-    private final InvoiceInternalApi invoiceApi;
     private final FailedPaymentRetryServiceScheduler failedPaymentRetryService;
     private final PluginFailureRetryServiceScheduler pluginFailureRetryService;
     private final AutoPayRetryServiceScheduler autoPayoffRetryService;
@@ -108,9 +107,8 @@ public class PaymentProcessor extends ProcessorBase {
                             final GlobalLocker locker,
                             final PaymentConfig paymentConfig,
                             @Named(PLUGIN_EXECUTOR_NAMED) final ExecutorService executor) {
-        super(pluginRegistry, accountUserApi, eventBus, paymentDao, tagUserApi, locker, executor);
+        super(pluginRegistry, accountUserApi, eventBus, paymentDao, tagUserApi, locker, executor, invoiceApi);
         this.paymentMethodProcessor = paymentMethodProcessor;
-        this.invoiceApi = invoiceApi;
         this.failedPaymentRetryService = failedPaymentRetryService;
         this.pluginFailureRetryService = pluginFailureRetryService;
         this.autoPayoffRetryService = autoPayoffRetryService;
@@ -250,9 +248,8 @@ public class PaymentProcessor extends ProcessorBase {
 
 
                                                                                                                 try {
-                                                                                                                    final Invoice invoice = invoiceApi.getInvoiceById(invoiceId, context);
-
-                                                                                                                    if (invoice.isMigrationInvoice()) {
+                                                                                                                    final Invoice invoice = rebalanceAndGetInvoice(account.getId(), invoiceId, context);
+                                                                                                                    if (invoice == null || invoice.isMigrationInvoice()) {
                                                                                                                         log.error("Received invoice for payment that is a migration invoice - don't know how to handle those yet: {}", invoice);
                                                                                                                         return null;
                                                                                                                     }
@@ -388,8 +385,8 @@ public class PaymentProcessor extends ProcessorBase {
                                                                                                            return null;
                                                                                                        }
 
-                                                                                                       final Invoice invoice = invoiceApi.getInvoiceById(payment.getInvoiceId(), context);
-                                                                                                       if (invoice.isMigrationInvoice()) {
+                                                                                                       final Invoice invoice = rebalanceAndGetInvoice(payment.getAccountId(), payment.getInvoiceId(), context);
+                                                                                                       if (invoice == null || invoice.isMigrationInvoice()) {
                                                                                                            return null;
                                                                                                        }
                                                                                                        if (invoice.getBalance().compareTo(BigDecimal.ZERO) <= 0) {

diff --git a/payment/src/main/java/com/ning/billing/payment/core/ProcessorBase.java b/payment/src/main/java/com/ning/billing/payment/core/ProcessorBase.java
index e3366aa..f6f23c3 100644
--- a/payment/src/main/java/com/ning/billing/payment/core/ProcessorBase.java
+++ b/payment/src/main/java/com/ning/billing/payment/core/ProcessorBase.java
@@ -30,6 +30,8 @@ import com.ning.billing.ErrorCode;
 import com.ning.billing.ObjectType;
 import com.ning.billing.account.api.Account;
 import com.ning.billing.account.api.AccountApiException;
+import com.ning.billing.invoice.api.Invoice;
+import com.ning.billing.invoice.api.InvoiceApiException;
 import com.ning.billing.osgi.api.OSGIServiceRegistration;
 import com.ning.billing.payment.api.PaymentApiException;
 import com.ning.billing.payment.dao.PaymentDao;
@@ -44,6 +46,7 @@ import com.ning.billing.util.globallocker.GlobalLocker;
 import com.ning.billing.util.globallocker.GlobalLocker.LockerType;
 import com.ning.billing.util.globallocker.LockFailedException;
 import com.ning.billing.util.svcapi.account.AccountInternalApi;
+import com.ning.billing.util.svcapi.invoice.InvoiceInternalApi;
 import com.ning.billing.util.svcapi.tag.TagInternalApi;
 import com.ning.billing.util.svcsapi.bus.InternalBus;
 import com.ning.billing.util.svcsapi.bus.InternalBus.EventBusException;
@@ -66,6 +69,7 @@ public abstract class ProcessorBase {
     protected final TagInternalApi tagInternalApi;
 
     private static final Logger log = LoggerFactory.getLogger(ProcessorBase.class);
+    protected final InvoiceInternalApi invoiceApi;
 
     public ProcessorBase(final OSGIServiceRegistration<PaymentPluginApi> pluginRegistry,
                          final AccountInternalApi accountInternalApi,
@@ -73,7 +77,7 @@ public abstract class ProcessorBase {
                          final PaymentDao paymentDao,
                          final TagInternalApi tagInternalApi,
                          final GlobalLocker locker,
-                         final ExecutorService executor) {
+                         final ExecutorService executor, final InvoiceInternalApi invoiceApi) {
         this.pluginRegistry = pluginRegistry;
         this.accountInternalApi = accountInternalApi;
         this.eventBus = eventBus;
@@ -81,6 +85,7 @@ public abstract class ProcessorBase {
         this.locker = locker;
         this.executor = executor;
         this.tagInternalApi = tagInternalApi;
+        this.invoiceApi = invoiceApi;
     }
 
     protected boolean isAccountAutoPayOff(final UUID accountId, final InternalTenantContext context) {
@@ -143,6 +148,12 @@ public abstract class ProcessorBase {
         }
     }
 
+    protected Invoice rebalanceAndGetInvoice(final UUID accountId, final UUID invoiceId, final InternalCallContext context) throws InvoiceApiException {
+        invoiceApi.consumeExistingCBAOnAccountWithUnpaidInvoices(accountId, context);
+        final Invoice invoice = invoiceApi.getInvoiceById(invoiceId, context);
+        return invoice;
+    }
+
     public interface WithAccountLockCallback<T> {
 
         public T doOperation() throws PaymentApiException;

diff --git a/payment/src/main/java/com/ning/billing/payment/core/RefundProcessor.java b/payment/src/main/java/com/ning/billing/payment/core/RefundProcessor.java
index 5a515d4..e76087d 100644
--- a/payment/src/main/java/com/ning/billing/payment/core/RefundProcessor.java
+++ b/payment/src/main/java/com/ning/billing/payment/core/RefundProcessor.java
@@ -73,7 +73,6 @@ public class RefundProcessor extends ProcessorBase {
 
     private static final Logger log = LoggerFactory.getLogger(RefundProcessor.class);
 
-    private final InvoiceInternalApi invoiceApi;
     private final InternalCallContextFactory internalCallContextFactory;
 
     @Inject
@@ -86,8 +85,7 @@ public class RefundProcessor extends ProcessorBase {
                            final PaymentDao paymentDao,
                            final GlobalLocker locker,
                            @Named(PLUGIN_EXECUTOR_NAMED) final ExecutorService executor) {
-        super(pluginRegistry, accountApi, eventBus, paymentDao, tagUserApi, locker, executor);
-        this.invoiceApi = invoiceApi;
+        super(pluginRegistry, accountApi, eventBus, paymentDao, tagUserApi, locker, executor, invoiceApi);
         this.internalCallContextFactory = internalCallContextFactory;
     }
 

diff --git a/util/src/main/java/com/ning/billing/util/svcapi/invoice/InvoiceInternalApi.java b/util/src/main/java/com/ning/billing/util/svcapi/invoice/InvoiceInternalApi.java
index ffcfaef..2e6095d 100644
--- a/util/src/main/java/com/ning/billing/util/svcapi/invoice/InvoiceInternalApi.java
+++ b/util/src/main/java/com/ning/billing/util/svcapi/invoice/InvoiceInternalApi.java
@@ -65,4 +65,12 @@ public interface InvoiceInternalApi {
     public InvoicePayment createRefund(UUID paymentId, BigDecimal amount, boolean isInvoiceAdjusted, final Map<UUID, BigDecimal> invoiceItemIdsWithAmounts,
                                        UUID paymentCookieId, InternalCallContext context) throws InvoiceApiException;
 
+    /**
+     * Rebalance CBA for account which have credit and unpaid invoices
+     *
+     * @param accountId account id
+     * @param context the callcontext
+     */
+    public void consumeExistingCBAOnAccountWithUnpaidInvoices(final UUID accountId, final InternalCallContext context) throws InvoiceApiException;
+
 }