killbill-aplcache

diff --git a/api/src/main/java/com/ning/billing/payment/api/CreditCardPaymentMethodInfo.java b/api/src/main/java/com/ning/billing/payment/api/CreditCardPaymentMethodInfo.java
index a9e25dc..572e362 100644
--- a/api/src/main/java/com/ning/billing/payment/api/CreditCardPaymentMethodInfo.java
+++ b/api/src/main/java/com/ning/billing/payment/api/CreditCardPaymentMethodInfo.java
@@ -192,4 +192,10 @@ public final class CreditCardPaymentMethodInfo extends PaymentMethodInfo {
     public String getMaskNumber() {
       return maskNumber;
     }
+
+    @Override
+    public String toString() {
+        return "CreditCardPaymentMethodInfo [cardHolderName=" + cardHolderName + ", cardType=" + cardType + ", expirationDate=" + expirationDate + ", maskNumber=" + maskNumber + ", cardAddress1=" + cardAddress1 + ", cardAddress2=" + cardAddress2 + ", cardCity=" + cardCity + ", cardState=" + cardState + ", cardPostalCode=" + cardPostalCode + ", cardCountry=" + cardCountry + "]";
+    }
+
 }

diff --git a/api/src/main/java/com/ning/billing/payment/api/PaymentApi.java b/api/src/main/java/com/ning/billing/payment/api/PaymentApi.java
index 58e9898..fd84927 100644
--- a/api/src/main/java/com/ning/billing/payment/api/PaymentApi.java
+++ b/api/src/main/java/com/ning/billing/payment/api/PaymentApi.java
@@ -17,6 +17,7 @@
 package com.ning.billing.payment.api;
 
 import java.util.List;
+import java.util.UUID;
 
 import javax.annotation.Nullable;
 
@@ -38,6 +39,7 @@ public interface PaymentApi {
 
     List<Either<PaymentError, PaymentInfo>> createPayment(String accountKey, List<String> invoiceIds);
     List<Either<PaymentError, PaymentInfo>> createPayment(Account account, List<String> invoiceIds);
+    Either<PaymentError, PaymentInfo> createPayment(UUID paymentAttemptId);
 
     List<Either<PaymentError, PaymentInfo>> createRefund(Account account, List<String> invoiceIds); //TODO
 
@@ -49,4 +51,10 @@ public interface PaymentApi {
 
     PaymentAttempt getPaymentAttemptForPaymentId(String id);
 
+    List<PaymentInfo> getPaymentInfo(List<String> invoiceIds);
+
+    PaymentAttempt getPaymentAttemptForInvoiceId(String invoiceId);
+
+    PaymentInfo getPaymentInfoForPaymentAttemptId(String paymentAttemptId);
+
 }

diff --git a/api/src/main/java/com/ning/billing/payment/api/PaymentAttempt.java b/api/src/main/java/com/ning/billing/payment/api/PaymentAttempt.java
index b5df043..fcccf9b 100644
--- a/api/src/main/java/com/ning/billing/payment/api/PaymentAttempt.java
+++ b/api/src/main/java/com/ning/billing/payment/api/PaymentAttempt.java
@@ -265,6 +265,7 @@ public class PaymentAttempt {
                                       createdDate,
                                       updatedDate);
         }
+
     }
 
     @Override

diff --git a/api/src/main/java/com/ning/billing/payment/api/PaymentService.java b/api/src/main/java/com/ning/billing/payment/api/PaymentService.java
index 988a00a..ede2506 100644
--- a/api/src/main/java/com/ning/billing/payment/api/PaymentService.java
+++ b/api/src/main/java/com/ning/billing/payment/api/PaymentService.java
@@ -23,4 +23,5 @@ public interface PaymentService extends KillbillService {
     String getName();
 
     PaymentApi getPaymentApi();
+
 }

diff --git a/invoice/src/main/java/com/ning/billing/invoice/glue/InvoiceModule.java b/invoice/src/main/java/com/ning/billing/invoice/glue/InvoiceModule.java
index 7c6e5cc..f6e6af0 100644
--- a/invoice/src/main/java/com/ning/billing/invoice/glue/InvoiceModule.java
+++ b/invoice/src/main/java/com/ning/billing/invoice/glue/InvoiceModule.java
@@ -66,8 +66,11 @@ public class InvoiceModule extends AbstractModule {
         bind(NextBillingDatePoster.class).to(DefaultNextBillingDatePoster.class).asEagerSingleton();
     }
 
-    protected void installInvoiceListener() {
+    protected void installGlobalLocker() {
         install(new GlobalLockerModule());
+    }
+
+    protected void installInvoiceListener() {
         bind(InvoiceListener.class).asEagerSingleton();
     }
 
@@ -81,5 +84,6 @@ public class InvoiceModule extends AbstractModule {
         installInvoiceDao();
         installInvoiceUserApi();
         installInvoicePaymentApi();
+        installGlobalLocker();
     }
 }

diff --git a/invoice/src/test/java/com/ning/billing/invoice/glue/InvoiceModuleWithMocks.java b/invoice/src/test/java/com/ning/billing/invoice/glue/InvoiceModuleWithMocks.java
index b180a03..01f0eb2 100644
--- a/invoice/src/test/java/com/ning/billing/invoice/glue/InvoiceModuleWithMocks.java
+++ b/invoice/src/test/java/com/ning/billing/invoice/glue/InvoiceModuleWithMocks.java
@@ -31,6 +31,11 @@ public class InvoiceModuleWithMocks extends InvoiceModule {
     }
 
     @Override
+    protected void installGlobalLocker() {
+        bind(GlobalLocker.class).to(MockGlobalLocker.class).asEagerSingleton();
+    }
+
+    @Override
     protected void installInvoiceListener() {
 
     }

diff --git a/payment/pom.xml b/payment/pom.xml
index bdb30c8..4a55289 100644
--- a/payment/pom.xml
+++ b/payment/pom.xml
@@ -62,7 +62,7 @@
         <dependency>
             <groupId>commons-io</groupId>
             <artifactId>commons-io</artifactId>
-            <version>2.0.1</version>
+            <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>commons-collections</groupId>
@@ -86,11 +86,6 @@
         <dependency>
             <groupId>com.ning.billing</groupId>
             <artifactId>killbill-util</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.ning.billing</groupId>
-            <artifactId>killbill-util</artifactId>
             <type>test-jar</type>
             <scope>test</scope>
         </dependency>

diff --git a/payment/src/main/java/com/ning/billing/payment/api/DefaultPaymentApi.java b/payment/src/main/java/com/ning/billing/payment/api/DefaultPaymentApi.java
index a8364bf..3b1a6f9 100644
--- a/payment/src/main/java/com/ning/billing/payment/api/DefaultPaymentApi.java
+++ b/payment/src/main/java/com/ning/billing/payment/api/DefaultPaymentApi.java
@@ -23,6 +23,9 @@ import java.util.UUID;
 
 import javax.annotation.Nullable;
 
+import org.apache.commons.lang.StringUtils;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,12 +38,14 @@ import com.ning.billing.invoice.model.DefaultInvoicePayment;
 import com.ning.billing.payment.dao.PaymentDao;
 import com.ning.billing.payment.provider.PaymentProviderPlugin;
 import com.ning.billing.payment.provider.PaymentProviderPluginRegistry;
+import com.ning.billing.payment.setup.PaymentConfig;
 
 public class DefaultPaymentApi implements PaymentApi {
     private final PaymentProviderPluginRegistry pluginRegistry;
     private final AccountUserApi accountUserApi;
     private final InvoicePaymentApi invoicePaymentApi;
     private final PaymentDao paymentDao;
+    private final PaymentConfig config;
 
     private static final Logger log = LoggerFactory.getLogger(DefaultPaymentApi.class);
 
@@ -48,11 +53,13 @@ public class DefaultPaymentApi implements PaymentApi {
     public DefaultPaymentApi(PaymentProviderPluginRegistry pluginRegistry,
                              AccountUserApi accountUserApi,
                              InvoicePaymentApi invoicePaymentApi,
-                             PaymentDao paymentDao) {
+                             PaymentDao paymentDao,
+                             PaymentConfig config) {
         this.pluginRegistry = pluginRegistry;
         this.accountUserApi = accountUserApi;
         this.invoicePaymentApi = invoicePaymentApi;
         this.paymentDao = paymentDao;
+        this.config = config;
     }
 
     @Override
@@ -127,6 +134,28 @@ public class DefaultPaymentApi implements PaymentApi {
     }
 
     @Override
+    public Either<PaymentError, PaymentInfo> createPayment(UUID paymentAttemptId) {
+        PaymentAttempt paymentAttempt = paymentDao.getPaymentAttemptById(paymentAttemptId);
+
+        if (paymentAttempt != null) {
+            Invoice invoice = invoicePaymentApi.getInvoice(paymentAttempt.getInvoiceId());
+            Account account = accountUserApi.getAccountById(paymentAttempt.getAccountId());
+
+            if (invoice != null && account != null) {
+                if (invoice.getBalance().compareTo(BigDecimal.ZERO) == 0 ) {
+                    // TODO: send a notification that invoice was ignored?
+                    log.info("Received invoice for payment with outstanding amount of 0 {} ", invoice);
+                    Either.left(new PaymentError("invoice_balance_0", "Invoice balance was 0"));
+                }
+                else {
+                    return processPayment(getPaymentProviderPlugin(account), account, invoice, paymentAttempt);
+                }
+            }
+        }
+        return Either.left(new PaymentError("retry_payment_error", "Could not load payment attempt, invoice or account for id " + paymentAttemptId));
+    }
+
+    @Override
     public List<Either<PaymentError, PaymentInfo>> createPayment(Account account, List<String> invoiceIds) {
         final PaymentProviderPlugin plugin = getPaymentProviderPlugin(account);
 
@@ -136,56 +165,93 @@ public class DefaultPaymentApi implements PaymentApi {
             Invoice invoice = invoicePaymentApi.getInvoice(UUID.fromString(invoiceId));
 
             if (invoice.getBalance().compareTo(BigDecimal.ZERO) == 0 ) {
-            // TODO: send a notification that invoice was ignored?
-                log.info("Received invoice for payment with outstanding amount of 0 {} ", invoice);
-            }
-            else if (invoiceId.equals(paymentDao.getPaymentAttemptForInvoiceId(invoiceId))) {
-                //TODO: do equals on invoice instead and only reject when invoice is exactly the same?
-                log.info("Duplicate invoice payment event, already received invoice {} ", invoice);
+                // TODO: send a notification that invoice was ignored?
+                log.info("Received invoice for payment with balance of 0 {} ", invoice);
+                Either.left(new PaymentError("invoice_balance_0", "Invoice balance was 0"));
             }
             else {
                 PaymentAttempt paymentAttempt = paymentDao.createPaymentAttempt(invoice);
-                Either<PaymentError, PaymentInfo> paymentOrError = plugin.processInvoice(account, invoice);
-                processedPaymentsOrErrors.add(paymentOrError);
 
-                PaymentInfo paymentInfo = null;
+                processedPaymentsOrErrors.add(processPayment(plugin, account, invoice, paymentAttempt));
+            }
+        }
 
-                if (paymentOrError.isRight()) {
-                    paymentInfo = paymentOrError.getRight();
-                    paymentDao.savePaymentInfo(paymentInfo);
+        return processedPaymentsOrErrors;
+    }
 
-                    Either<PaymentError, PaymentMethodInfo> paymentMethodInfoOrError = plugin.getPaymentMethodInfo(paymentInfo.getPaymentMethodId());
+    private Either<PaymentError, PaymentInfo> processPayment(PaymentProviderPlugin plugin, Account account, Invoice invoice, PaymentAttempt paymentAttempt) {
+        Either<PaymentError, PaymentInfo> paymentOrError = plugin.processInvoice(account, invoice);
+        PaymentInfo paymentInfo = null;
 
-                    if (paymentMethodInfoOrError.isRight()) {
-                        PaymentMethodInfo paymentMethodInfo = paymentMethodInfoOrError.getRight();
+        if (paymentOrError.isLeft()) {
+            String error = StringUtils.substring(paymentOrError.getLeft().getMessage() + paymentOrError.getLeft().getType(), 0, 100);
+            log.info("Could not process a payment for " + paymentAttempt + " error was " + error);
 
-                        if (paymentMethodInfo instanceof CreditCardPaymentMethodInfo) {
-                            CreditCardPaymentMethodInfo ccPaymentMethod = (CreditCardPaymentMethodInfo)paymentMethodInfo;
-                            paymentDao.updatePaymentInfo(ccPaymentMethod.getType(), paymentInfo.getPaymentId(), ccPaymentMethod.getCardType(), ccPaymentMethod.getCardCountry());
-                        }
-                        else if (paymentMethodInfo instanceof PaypalPaymentMethodInfo) {
-                            PaypalPaymentMethodInfo paypalPaymentMethodInfo = (PaypalPaymentMethodInfo)paymentMethodInfo;
-                            paymentDao.updatePaymentInfo(paypalPaymentMethodInfo.getType(), paymentInfo.getPaymentId(), null, null);
-                        }
-                    }
+            scheduleRetry(paymentAttempt, error);
+        }
+        else {
+            paymentInfo = paymentOrError.getRight();
+            paymentDao.savePaymentInfo(paymentInfo);
 
+            Either<PaymentError, PaymentMethodInfo> paymentMethodInfoOrError = plugin.getPaymentMethodInfo(paymentInfo.getPaymentMethodId());
 
-                    if (paymentInfo.getPaymentId() != null) {
-                        paymentDao.updatePaymentAttemptWithPaymentId(paymentAttempt.getPaymentAttemptId(), paymentInfo.getPaymentId());
-                    }
-                }
+            if (paymentMethodInfoOrError.isRight()) {
+                PaymentMethodInfo paymentMethodInfo = paymentMethodInfoOrError.getRight();
 
-                invoicePaymentApi.notifyOfPaymentAttempt(new DefaultInvoicePayment(paymentAttempt.getPaymentAttemptId(),
-                                                                                   invoice.getId(),
-                                                                                   paymentAttempt.getPaymentAttemptDate(),
-                                                                                   paymentInfo == null || paymentInfo.getStatus().equalsIgnoreCase("Error") ? null : paymentInfo.getAmount(),
-//                                                                                 paymentInfo.getRefundAmount(), TODO
-                                                                                   paymentInfo == null || paymentInfo.getStatus().equalsIgnoreCase("Error") ? null : invoice.getCurrency()));
+                if (paymentMethodInfo instanceof CreditCardPaymentMethodInfo) {
+                    CreditCardPaymentMethodInfo ccPaymentMethod = (CreditCardPaymentMethodInfo)paymentMethodInfo;
+                    paymentDao.updatePaymentInfo(ccPaymentMethod.getType(), paymentInfo.getPaymentId(), ccPaymentMethod.getCardType(), ccPaymentMethod.getCardCountry());
+                }
+                else if (paymentMethodInfo instanceof PaypalPaymentMethodInfo) {
+                    PaypalPaymentMethodInfo paypalPaymentMethodInfo = (PaypalPaymentMethodInfo)paymentMethodInfo;
+                    paymentDao.updatePaymentInfo(paypalPaymentMethodInfo.getType(), paymentInfo.getPaymentId(), null, null);
+                }
+            }
 
+            if (paymentInfo.getPaymentId() != null) {
+                paymentDao.updatePaymentAttemptWithPaymentId(paymentAttempt.getPaymentAttemptId(), paymentInfo.getPaymentId());
             }
         }
 
-        return processedPaymentsOrErrors;
+        invoicePaymentApi.notifyOfPaymentAttempt(new DefaultInvoicePayment(paymentAttempt.getPaymentAttemptId(),
+                                                                           invoice.getId(),
+                                                                           paymentAttempt.getPaymentAttemptDate(),
+                                                                           paymentInfo == null || paymentInfo.getStatus().equalsIgnoreCase("Error") ? null : paymentInfo.getAmount(),
+//                                                                         paymentInfo.getRefundAmount(), TODO
+                                                                           paymentInfo == null || paymentInfo.getStatus().equalsIgnoreCase("Error") ? null : invoice.getCurrency()));
+
+        return paymentOrError;
+    }
+
+    private void scheduleRetry(PaymentAttempt paymentAttempt, String error) {
+        final List<Integer> retryDays = config.getPaymentRetryDays();
+
+        int retryCount = 0;
+
+        if (paymentAttempt.getRetryCount() != null) {
+            retryCount = paymentAttempt.getRetryCount();
+        }
+
+        if (retryCount < retryDays.size()) {
+            int retryInDays = 0;
+            DateTime nextRetryDate = new DateTime(DateTimeZone.UTC);
+
+            try {
+                retryInDays = retryDays.get(retryCount);
+                nextRetryDate = nextRetryDate.plusDays(retryInDays);
+            }
+            catch (NumberFormatException ex) {
+                log.error("Could not get retry day for retry count {}", retryCount);
+            }
+
+            paymentDao.updatePaymentAttemptWithRetryInfo(paymentAttempt.getPaymentAttemptId(), retryCount + 1, nextRetryDate);
+        }
+        else if (retryCount == retryDays.size()) {
+            log.info("Last payment retry failed for {} ", paymentAttempt);
+        }
+        else {
+            log.error("Cannot update payment retry information because retry count is invalid {} ", retryCount);
+        }
     }
 
     @Override
@@ -212,4 +278,19 @@ public class DefaultPaymentApi implements PaymentApi {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public List<PaymentInfo> getPaymentInfo(List<String> invoiceIds) {
+        return paymentDao.getPaymentInfo(invoiceIds);
+    }
+
+    @Override
+    public PaymentAttempt getPaymentAttemptForInvoiceId(String invoiceId) {
+        return paymentDao.getPaymentAttemptForInvoiceId(invoiceId);
+    }
+
+    @Override
+    public PaymentInfo getPaymentInfoForPaymentAttemptId(String paymentAttemptId) {
+        return paymentDao.getPaymentInfoForPaymentAttemptId(paymentAttemptId);
+    }
+
 }

diff --git a/payment/src/main/java/com/ning/billing/payment/dao/DefaultPaymentDao.java b/payment/src/main/java/com/ning/billing/payment/dao/DefaultPaymentDao.java
index eacc226..a86db13 100644
--- a/payment/src/main/java/com/ning/billing/payment/dao/DefaultPaymentDao.java
+++ b/payment/src/main/java/com/ning/billing/payment/dao/DefaultPaymentDao.java
@@ -16,8 +16,10 @@
 
 package com.ning.billing.payment.dao;
 
+import java.util.List;
 import java.util.UUID;
 
+import org.joda.time.DateTime;
 import org.skife.jdbi.v2.IDBI;
 
 import com.google.inject.Inject;
@@ -44,6 +46,12 @@ public class DefaultPaymentDao implements PaymentDao {
     }
 
     @Override
+    public PaymentAttempt createPaymentAttempt(PaymentAttempt paymentAttempt) {
+        sqlDao.insertPaymentAttempt(paymentAttempt);
+        return paymentAttempt;
+    }
+
+    @Override
     public PaymentAttempt createPaymentAttempt(Invoice invoice) {
         final PaymentAttempt paymentAttempt = new PaymentAttempt(UUID.randomUUID(), invoice);
 
@@ -66,4 +74,29 @@ public class DefaultPaymentDao implements PaymentDao {
         sqlDao.updatePaymentInfo(type, paymentId, cardType, cardCountry);
     }
 
+    @Override
+    public List<PaymentInfo> getPaymentInfo(List<String> invoiceIds) {
+        return sqlDao.getPaymentInfos(invoiceIds);
+    }
+
+    @Override
+    public List<PaymentAttempt> getPaymentAttemptsForInvoiceIds(List<String> invoiceIds) {
+        return sqlDao.getPaymentAttemptsForInvoiceIds(invoiceIds);
+    }
+
+    @Override
+    public void updatePaymentAttemptWithRetryInfo(UUID paymentAttemptId, int retryCount, DateTime nextRetryDate) {
+        sqlDao.updatePaymentAttemptWithRetryInfo(paymentAttemptId.toString(), retryCount, nextRetryDate);
+    }
+
+    @Override
+    public PaymentAttempt getPaymentAttemptById(UUID paymentAttemptId) {
+        return sqlDao.getPaymentAttemptById(paymentAttemptId.toString());
+    }
+
+    @Override
+    public PaymentInfo getPaymentInfoForPaymentAttemptId(String paymentAttemptIdStr) {
+        return sqlDao.getPaymentInfoForPaymentAttemptId(paymentAttemptIdStr);
+    }
+
 }

diff --git a/payment/src/main/java/com/ning/billing/payment/dao/PaymentDao.java b/payment/src/main/java/com/ning/billing/payment/dao/PaymentDao.java
index 5cf065b..de2d8cc 100644
--- a/payment/src/main/java/com/ning/billing/payment/dao/PaymentDao.java
+++ b/payment/src/main/java/com/ning/billing/payment/dao/PaymentDao.java
@@ -16,8 +16,11 @@
 
 package com.ning.billing.payment.dao;
 
+import java.util.List;
 import java.util.UUID;
 
+import org.joda.time.DateTime;
+
 import com.ning.billing.invoice.api.Invoice;
 import com.ning.billing.payment.api.PaymentAttempt;
 import com.ning.billing.payment.api.PaymentInfo;
@@ -25,15 +28,23 @@ import com.ning.billing.payment.api.PaymentInfo;
 public interface PaymentDao {
 
     PaymentAttempt createPaymentAttempt(Invoice invoice);
+    PaymentAttempt createPaymentAttempt(PaymentAttempt paymentAttempt);
 
     void savePaymentInfo(PaymentInfo right);
 
     PaymentAttempt getPaymentAttemptForPaymentId(String paymentId);
+    List<PaymentAttempt> getPaymentAttemptsForInvoiceIds(List<String> invoiceIds);
 
     void updatePaymentAttemptWithPaymentId(UUID paymentAttemptId, String paymentId);
+    void updatePaymentAttemptWithRetryInfo(UUID paymentAttemptId, int retryCount, DateTime nextRetryDate);
 
     PaymentAttempt getPaymentAttemptForInvoiceId(String invoiceId);
 
     void updatePaymentInfo(String paymentMethodType, String paymentId, String cardType, String cardCountry);
 
+    List<PaymentInfo> getPaymentInfo(List<String> invoiceIds);
+
+    PaymentAttempt getPaymentAttemptById(UUID paymentAttemptId);
+    PaymentInfo getPaymentInfoForPaymentAttemptId(String paymentAttemptId);
+
 }

diff --git a/payment/src/main/java/com/ning/billing/payment/dao/PaymentSqlDao.java b/payment/src/main/java/com/ning/billing/payment/dao/PaymentSqlDao.java
index 972ee64..9919d34 100644
--- a/payment/src/main/java/com/ning/billing/payment/dao/PaymentSqlDao.java
+++ b/payment/src/main/java/com/ning/billing/payment/dao/PaymentSqlDao.java
@@ -21,6 +21,7 @@ import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Timestamp;
 import java.util.Date;
+import java.util.List;
 import java.util.UUID;
 
 import org.joda.time.DateTime;
@@ -37,6 +38,7 @@ import org.skife.jdbi.v2.sqlobject.mixins.Transactional;
 import org.skife.jdbi.v2.sqlobject.mixins.Transmogrifier;
 import org.skife.jdbi.v2.sqlobject.stringtemplate.ExternalizedSqlViaStringTemplate3;
 import org.skife.jdbi.v2.tweak.ResultSetMapper;
+import org.skife.jdbi.v2.unstable.BindIn;
 
 import com.ning.billing.catalog.api.Currency;
 import com.ning.billing.payment.api.PaymentAttempt;
@@ -53,18 +55,39 @@ public interface PaymentSqlDao extends Transactional<PaymentSqlDao>, CloseMe, Tr
 
     @SqlQuery
     @Mapper(PaymentAttemptMapper.class)
+    PaymentAttempt getPaymentAttemptById(@Bind("payment_attempt_id") String paymentAttemptId);
+
+    @SqlQuery
+    @Mapper(PaymentAttemptMapper.class)
     PaymentAttempt getPaymentAttemptForInvoiceId(@Bind("invoice_id") String invoiceId);
 
+    @SqlQuery
+    @Mapper(PaymentAttemptMapper.class)
+    List<PaymentAttempt> getPaymentAttemptsForInvoiceIds(@BindIn("invoiceIds") List<String> invoiceIds);
+
+    @SqlQuery
+    @Mapper(PaymentInfoMapper.class)
+    PaymentInfo getPaymentInfoForPaymentAttemptId(@Bind("payment_attempt_id") String paymentAttemptId);
+
     @SqlUpdate
     void updatePaymentAttemptWithPaymentId(@Bind("payment_attempt_id") String paymentAttemptId,
                                            @Bind("payment_id") String paymentId);
 
     @SqlUpdate
+    void updatePaymentAttemptWithRetryInfo(@Bind("payment_attempt_id") String paymentAttemptId,
+                                           @Bind("retry_count") int retryCount,
+                                           @Bind("next_retry_dt") DateTime nextRetryDate);
+
+    @SqlUpdate
     void updatePaymentInfo(@Bind("payment_method") String paymentMethod,
                            @Bind("payment_id") String paymentId,
                            @Bind("card_type") String cardType,
                            @Bind("card_country") String cardCountry);
 
+    @SqlQuery
+    @Mapper(PaymentInfoMapper.class)
+    List<PaymentInfo> getPaymentInfos(@BindIn("invoiceIds") List<String> invoiceIds);
+
     @SqlUpdate
     void insertPaymentInfo(@Bind(binder = PaymentInfoBinder.class) PaymentInfo paymentInfo);
 

diff --git a/payment/src/main/java/com/ning/billing/payment/RequestProcessor.java b/payment/src/main/java/com/ning/billing/payment/RequestProcessor.java
index 892d424..d53c236 100644
--- a/payment/src/main/java/com/ning/billing/payment/RequestProcessor.java
+++ b/payment/src/main/java/com/ning/billing/payment/RequestProcessor.java
@@ -31,7 +31,6 @@ import com.ning.billing.payment.api.Either;
 import com.ning.billing.payment.api.PaymentApi;
 import com.ning.billing.payment.api.PaymentError;
 import com.ning.billing.payment.api.PaymentInfo;
-import com.ning.billing.payment.provider.PaymentProviderPlugin;
 import com.ning.billing.payment.provider.PaymentProviderPluginRegistry;
 import com.ning.billing.util.bus.Bus;
 import com.ning.billing.util.bus.Bus.EventBusException;
@@ -40,7 +39,6 @@ public class RequestProcessor {
     public static final String PAYMENT_PROVIDER_KEY = "paymentProvider";
     private final AccountUserApi accountUserApi;
     private final PaymentApi paymentApi;
-    private final PaymentProviderPluginRegistry pluginRegistry;
     private final Bus eventBus;
 
     private static final Logger log = LoggerFactory.getLogger(RequestProcessor.class);
@@ -52,7 +50,6 @@ public class RequestProcessor {
                             Bus eventBus) {
         this.accountUserApi = accountUserApi;
         this.paymentApi = paymentApi;
-        this.pluginRegistry = pluginRegistry;
         this.eventBus = eventBus;
     }
 
@@ -77,20 +74,4 @@ public class RequestProcessor {
             throw new RuntimeException(ex);
         }
     }
-
-    @Subscribe
-    public void receivePaymentInfoRequest(PaymentInfoRequest paymentInfoRequest) throws EventBusException {
-        final Account account = accountUserApi.getAccountById(paymentInfoRequest.getAccountId());
-        if (account == null) {
-            log.info("could not process payment info request: could not find a valid account for event {}", paymentInfoRequest);
-        }
-        else {
-            final String paymentProviderName = account.getFieldValue(PAYMENT_PROVIDER_KEY);
-            final PaymentProviderPlugin plugin = pluginRegistry.getPlugin(paymentProviderName);
-
-            Either<PaymentError, PaymentInfo> result = plugin.getPaymentInfo(paymentInfoRequest.getPaymentId());
-
-            eventBus.post(result.isLeft() ? result.getLeft() : result.getRight());
-        }
-    }
 }

diff --git a/payment/src/main/java/com/ning/billing/payment/RetryService.java b/payment/src/main/java/com/ning/billing/payment/RetryService.java
new file mode 100644
index 0000000..d6db526
--- /dev/null
+++ b/payment/src/main/java/com/ning/billing/payment/RetryService.java
@@ -0,0 +1,103 @@
+/*
+ * Copyright 2010-2012 Ning, Inc.
+ *
+ * Ning 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 com.ning.billing.payment;
+
+import java.util.UUID;
+
+import org.joda.time.DateTime;
+import org.skife.jdbi.v2.sqlobject.mixins.Transmogrifier;
+
+import com.google.inject.Inject;
+import com.ning.billing.lifecycle.KillbillService;
+import com.ning.billing.lifecycle.LifecycleHandlerType;
+import com.ning.billing.lifecycle.LifecycleHandlerType.LifecycleLevel;
+import com.ning.billing.payment.api.PaymentApi;
+import com.ning.billing.payment.api.PaymentAttempt;
+import com.ning.billing.payment.api.PaymentInfo;
+import com.ning.billing.payment.setup.PaymentConfig;
+import com.ning.billing.util.notificationq.NotificationKey;
+import com.ning.billing.util.notificationq.NotificationQueue;
+import com.ning.billing.util.notificationq.NotificationQueueService;
+import com.ning.billing.util.notificationq.NotificationQueueService.NotificationQueueAlreadyExists;
+import com.ning.billing.util.notificationq.NotificationQueueService.NotificationQueueHandler;
+
+public class RetryService implements KillbillService {
+    public static final String SERVICE_NAME = "retry-service";
+    public static final String QUEUE_NAME = "retry-events";
+
+    private final NotificationQueueService notificationQueueService;
+    private final PaymentConfig config;
+    private final PaymentApi paymentApi;
+    private NotificationQueue retryQueue;
+
+    @Inject
+    public RetryService(NotificationQueueService notificationQueueService,
+                        PaymentConfig config,
+                        PaymentApi paymentApi) {
+        this.notificationQueueService = notificationQueueService;
+        this.paymentApi = paymentApi;
+        this.config = config;
+    }
+
+    @Override
+    public String getName() {
+        return SERVICE_NAME;
+    }
+
+    @LifecycleHandlerType(LifecycleLevel.INIT_SERVICE)
+    public void initialize() throws NotificationQueueAlreadyExists {
+        retryQueue = notificationQueueService.createNotificationQueue(SERVICE_NAME, QUEUE_NAME, new NotificationQueueHandler() {
+            @Override
+            public void handleReadyNotification(String notificationKey, DateTime eventDateTime) {
+                retry(notificationKey);
+            }
+        },
+        config);
+    }
+
+    @LifecycleHandlerType(LifecycleLevel.START_SERVICE)
+    public void start() {
+        retryQueue.startQueue();
+    }
+
+    @LifecycleHandlerType(LifecycleLevel.STOP_SERVICE)
+    public void stop() {
+        if (retryQueue != null) {
+            retryQueue.stopQueue();
+         }
+    }
+
+    public void scheduleRetry(Transmogrifier transactionalDao, PaymentAttempt paymentAttempt, DateTime timeOfRetry) {
+        final String id = paymentAttempt.getPaymentAttemptId().toString();
+
+        NotificationKey key = new NotificationKey() {
+            @Override
+            public String toString() {
+                return id;
+            }
+        };
+        retryQueue.recordFutureNotificationFromTransaction(transactionalDao, timeOfRetry, key);
+    }
+
+    private void retry(String paymentAttemptId) {
+        PaymentInfo paymentInfo = paymentApi.getPaymentInfoForPaymentAttemptId(paymentAttemptId);
+
+        if (paymentInfo != null && !"Processed".equalsIgnoreCase(paymentInfo.getStatus())) {
+            paymentApi.createPayment(UUID.fromString(paymentAttemptId));
+        }
+    }
+}

diff --git a/payment/src/main/java/com/ning/billing/payment/setup/PaymentConfig.java b/payment/src/main/java/com/ning/billing/payment/setup/PaymentConfig.java
index cdc5384..46f00fc 100644
--- a/payment/src/main/java/com/ning/billing/payment/setup/PaymentConfig.java
+++ b/payment/src/main/java/com/ning/billing/payment/setup/PaymentConfig.java
@@ -16,11 +16,37 @@
 
 package com.ning.billing.payment.setup;
 
+import java.util.List;
+
 import org.skife.config.Config;
-import org.skife.config.DefaultNull;
+import org.skife.config.Default;
+
+import com.ning.billing.util.notificationq.NotificationConfig;
 
-public interface PaymentConfig {
+public interface PaymentConfig extends NotificationConfig {
     @Config("killbill.payment.provider.default")
-    @DefaultNull
+    @Default("noop")
     public String getDefaultPaymentProvider();
+
+    @Config("killbill.payment.retry.days")
+    @Default("8,8,8")
+    public List<Integer> getPaymentRetryDays();
+
+    @Config("killbill.payment.dao.claim.time")
+    @Default("60000")
+    public long getDaoClaimTimeMs();
+
+    @Config("killbill.payment.dao.ready.max")
+    @Default("10")
+    public int getDaoMaxReadyEvents();
+
+    @Config("killbill.payment.engine.notifications.sleep")
+    @Default("500")
+    public long getNotificationSleepTimeMs();
+
+    @Config("killbill.payment.engine.events.off")
+    // turn off payment retries by default
+    @Default("true")
+    public boolean isNotificationProcessingOff();
+
 }

diff --git a/payment/src/main/java/com/ning/billing/payment/setup/PaymentModule.java b/payment/src/main/java/com/ning/billing/payment/setup/PaymentModule.java
index 935b968..85641ed 100644
--- a/payment/src/main/java/com/ning/billing/payment/setup/PaymentModule.java
+++ b/payment/src/main/java/com/ning/billing/payment/setup/PaymentModule.java
@@ -22,6 +22,7 @@ import org.skife.config.ConfigurationObjectFactory;
 
 import com.google.inject.AbstractModule;
 import com.ning.billing.payment.RequestProcessor;
+import com.ning.billing.payment.RetryService;
 import com.ning.billing.payment.api.DefaultPaymentApi;
 import com.ning.billing.payment.api.PaymentApi;
 import com.ning.billing.payment.api.PaymentService;
@@ -47,6 +48,10 @@ public class PaymentModule extends AbstractModule {
     protected void installPaymentProviderPlugins(PaymentConfig config) {
     }
 
+    protected void installRetryEngine() {
+        bind(RetryService.class).asEagerSingleton();
+    }
+
     @Override
     protected void configure() {
         final ConfigurationObjectFactory factory = new ConfigurationObjectFactory(props);

diff --git a/payment/src/main/resources/com/ning/billing/payment/dao/PaymentSqlDao.sql.stg b/payment/src/main/resources/com/ning/billing/payment/dao/PaymentSqlDao.sql.stg
index a62c366..6576ecb 100644
--- a/payment/src/main/resources/com/ning/billing/payment/dao/PaymentSqlDao.sql.stg
+++ b/payment/src/main/resources/com/ning/billing/payment/dao/PaymentSqlDao.sql.stg
@@ -24,6 +24,7 @@ paymentInfoFields(prefix) ::= <<
     <prefix>payment_type,
     <prefix>status,
     <prefix>reference_id,
+    <prefix>payment_method_id,
     <prefix>payment_method,
     <prefix>card_type,
     <prefix>card_country,
@@ -43,6 +44,18 @@ getPaymentAttemptForPaymentId() ::= <<
      WHERE payment_id = :payment_id
 >>
 
+getPaymentAttemptById() ::= <<
+    SELECT <paymentAttemptFields()>
+      FROM payment_attempts
+     WHERE payment_attempt_id = :payment_attempt_id
+>>
+
+getPaymentAttemptsForInvoiceIds(invoiceIds) ::= <<
+    SELECT <paymentAttemptFields()>
+      FROM payment_attempts
+     WHERE invoice_id in (<invoiceIds>)
+>>
+
 getPaymentAttemptForInvoiceId() ::= <<
     SELECT <paymentAttemptFields()>
       FROM payment_attempts
@@ -58,7 +71,7 @@ updatePaymentAttemptWithPaymentId() ::= <<
 
 insertPaymentInfo() ::= <<
     INSERT INTO payments (<paymentInfoFields()>)
-    VALUES (:payment_id, :amount, :refund_amount, :bank_identification_number, :payment_number, :payment_type, :status, :reference_id, :payment_method, :card_type, :card_country, :effective_dt, :created_dt, :updated_dt);
+    VALUES (:payment_id, :amount, :refund_amount, :bank_identification_number, :payment_number, :payment_type, :status, :reference_id, :payment_method_id, :payment_method, :card_type, :card_country, :effective_dt, :created_dt, :updated_dt);
 >>
 
 updatePaymentInfo() ::= <<
@@ -68,4 +81,19 @@ updatePaymentInfo() ::= <<
            card_country = :card_country,
            updated_dt = NOW()
      WHERE payment_id = :payment_id
+>>
+
+updatePaymentAttemptWithRetryInfo() ::= <<
+    UPDATE payment_attempts
+       SET retry_count = :retry_count,
+           next_retry_dt = :next_retry_dt,
+           updated_dt = NOW()
+     WHERE payment_attempt_id = :payment_attempt_id
+>>
+
+getPaymentInfos(invoiceIds) ::= <<
+    SELECT <paymentInfoFields("p.")>
+      FROM payments p, payment_attempts pa
+     WHERE pa.invoice_id in (<invoiceIds>)
+       AND pa.payment_id = p.payment_id
 >>
\ No newline at end of file

diff --git a/payment/src/test/java/com/ning/billing/payment/api/TestPaymentApi.java b/payment/src/test/java/com/ning/billing/payment/api/TestPaymentApi.java
index 8c9d623..2ef26e4 100644
--- a/payment/src/test/java/com/ning/billing/payment/api/TestPaymentApi.java
+++ b/payment/src/test/java/com/ning/billing/payment/api/TestPaymentApi.java
@@ -100,9 +100,19 @@ public abstract class TestPaymentApi {
         assertEquals(paymentAttempt.getPaymentId(), paymentInfo.getPaymentId());
         assertEquals(paymentAttempt.getPaymentAttemptDate().withMillisOfSecond(0).withSecondOfMinute(0), now.withMillisOfSecond(0).withSecondOfMinute(0));
 
+        List<PaymentInfo> paymentInfos = paymentApi.getPaymentInfo(Arrays.asList(invoice.getId().toString()));
+        assertNotNull(paymentInfos);
+        assertTrue(paymentInfos.size() > 0);
+
+        PaymentInfo paymentInfoFromGet = paymentInfos.get(0);
+        assertEquals(paymentInfo, paymentInfoFromGet);
+
+        PaymentAttempt paymentAttemptFromGet = paymentApi.getPaymentAttemptForInvoiceId(invoice.getId().toString());
+        assertEquals(paymentAttempt, paymentAttemptFromGet);
+
     }
 
-    private PaymentProviderAccount setupAccountWithPaymentMethod() throws AccountApiException {
+    private PaymentProviderAccount setupAccountWithPaypalPaymentMethod() throws AccountApiException {
         final Account account = testHelper.createTestPayPalAccount();
         paymentApi.createPaymentProviderAccount(account);
 
@@ -131,9 +141,10 @@ public abstract class TestPaymentApi {
     }
 
     @Test(enabled=true)
-    public void testCreatePaymentMethod() throws AccountApiException {
-        PaymentProviderAccount account = setupAccountWithPaymentMethod();
+    public void testCreatePaypalPaymentMethod() throws AccountApiException {
+        PaymentProviderAccount account = setupAccountWithPaypalPaymentMethod();
         assertNotNull(account);
+        Either<PaymentError, List<PaymentMethodInfo>> paymentMethodsOrError = paymentApi.getPaymentMethods(account.getAccountKey());
     }
 
     @Test(enabled=true)
@@ -160,7 +171,7 @@ public abstract class TestPaymentApi {
 
     @Test(enabled=true)
     public void testCannotDeleteDefaultPaymentMethod() throws AccountApiException {
-        PaymentProviderAccount account = setupAccountWithPaymentMethod();
+        PaymentProviderAccount account = setupAccountWithPaypalPaymentMethod();
 
         Either<PaymentError, Void> errorOrVoid = paymentApi.deletePaymentMethod(account.getAccountKey(), account.getDefaultPaymentMethodId());
 
@@ -197,4 +208,5 @@ public abstract class TestPaymentApi {
         assertTrue(errorOrVoid1.isRight());
         assertTrue(errorOrVoid2.isLeft());
     }
+
 }

diff --git a/payment/src/test/java/com/ning/billing/payment/dao/MockPaymentDao.java b/payment/src/test/java/com/ning/billing/payment/dao/MockPaymentDao.java
index bb83927..b245e55 100644
--- a/payment/src/test/java/com/ning/billing/payment/dao/MockPaymentDao.java
+++ b/payment/src/test/java/com/ning/billing/payment/dao/MockPaymentDao.java
@@ -16,10 +16,16 @@
 
 package com.ning.billing.payment.dao;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.joda.time.DateTime;
+
+import com.google.common.base.Predicate;
+import com.google.common.collect.Collections2;
 import com.ning.billing.invoice.api.Invoice;
 import com.ning.billing.payment.api.PaymentAttempt;
 import com.ning.billing.payment.api.PaymentInfo;
@@ -46,6 +52,12 @@ public class MockPaymentDao implements PaymentDao {
     }
 
     @Override
+    public PaymentAttempt createPaymentAttempt(PaymentAttempt paymentAttempt) {
+        paymentAttempts.put(paymentAttempt.getPaymentAttemptId(), paymentAttempt);
+        return paymentAttempt;
+    }
+
+    @Override
     public void savePaymentInfo(PaymentInfo paymentInfo) {
         payments.put(paymentInfo.getPaymentId(), paymentInfo);
     }
@@ -63,7 +75,7 @@ public class MockPaymentDao implements PaymentDao {
     @Override
     public PaymentAttempt getPaymentAttemptForInvoiceId(String invoiceId) {
         for (PaymentAttempt paymentAttempt : paymentAttempts.values()) {
-            if (invoiceId.equals(paymentAttempt.getInvoiceId())) {
+            if (invoiceId.equals(paymentAttempt.getInvoiceId().toString())) {
                 return paymentAttempt;
             }
         }
@@ -72,8 +84,59 @@ public class MockPaymentDao implements PaymentDao {
 
     @Override
     public void updatePaymentInfo(String paymentMethodType, String paymentId, String cardType, String cardCountry) {
-        // TODO Auto-generated method stub
+        PaymentInfo existingPayment = payments.get(paymentId);
+        if (existingPayment != null) {
+            PaymentInfo payment = existingPayment.cloner().setPaymentMethod(paymentMethodType).setCardType(cardType).setCardCountry(cardCountry).build();
+            payments.put(paymentId, payment);
+        }
+    }
+
+    @Override
+    public List<PaymentInfo> getPaymentInfo(List<String> invoiceIds) {
+        List<PaymentAttempt> attempts = getPaymentAttemptsForInvoiceIds(invoiceIds);
+        List<PaymentInfo> paymentsToReturn = new ArrayList<PaymentInfo>(invoiceIds.size());
 
+        for (final PaymentAttempt attempt : attempts) {
+            paymentsToReturn.addAll(Collections2.filter(payments.values(), new Predicate<PaymentInfo>() {
+                @Override
+                public boolean apply(PaymentInfo input) {
+                    return input.getPaymentId().equals(attempt.getPaymentId());
+                }
+            }));
+        }
+        return paymentsToReturn;
+    }
+
+    @Override
+    public List<PaymentAttempt> getPaymentAttemptsForInvoiceIds(List<String> invoiceIds) {
+        List<PaymentAttempt> paymentAttempts = new ArrayList<PaymentAttempt>(invoiceIds.size());
+        for (String invoiceId : invoiceIds) {
+            PaymentAttempt attempt = getPaymentAttemptForInvoiceId(invoiceId);
+            if (attempt != null) {
+                paymentAttempts.add(attempt);
+            }
+        }
+        return paymentAttempts;
+    }
+
+    @Override
+    public void updatePaymentAttemptWithRetryInfo(UUID paymentAttemptId, int retryCount, DateTime nextRetryDate) {
+        PaymentAttempt existingAttempt = paymentAttempts.get(paymentAttemptId);
+        if (existingAttempt != null) {
+            PaymentAttempt attempt = existingAttempt.cloner().setPaymentAttemptId(paymentAttemptId).setRetryCount(retryCount).setNextRetryDate(nextRetryDate).build();
+            paymentAttempts.put(paymentAttemptId, attempt);
+        }
+    }
+
+    @Override
+    public PaymentAttempt getPaymentAttemptById(UUID paymentAttemptId) {
+        return paymentAttempts.get(paymentAttemptId);
+    }
+
+    @Override
+    public PaymentInfo getPaymentInfoForPaymentAttemptId(String paymentAttemptId) {
+        // TODO Auto-generated method stub
+        return null;
     }
 
 }

diff --git a/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDao.java b/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDao.java
index 6c57c77..18b0a15 100644
--- a/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDao.java
+++ b/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDao.java
@@ -17,17 +17,22 @@
 package com.ning.billing.payment.dao;
 
 import java.math.BigDecimal;
+import java.util.Arrays;
 import java.util.UUID;
 
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
+import org.testng.Assert;
 import org.testng.annotations.Test;
 
+import com.ning.billing.account.api.AccountApiException;
+import com.ning.billing.catalog.api.Currency;
+import com.ning.billing.payment.api.PaymentAttempt;
 import com.ning.billing.payment.api.PaymentInfo;
 
 public abstract class TestPaymentDao {
 
-    protected PaymentDao dao;
+    protected PaymentDao paymentDao;
 
     @Test
     public void testCreatePayment() {
@@ -44,7 +49,7 @@ public abstract class TestPaymentDao {
                                                            .setEffectiveDate(new DateTime(DateTimeZone.UTC))
                                                            .build();
 
-        dao.savePaymentInfo(paymentInfo);
+        paymentDao.savePaymentInfo(paymentInfo);
     }
 
     @Test
@@ -62,10 +67,55 @@ public abstract class TestPaymentDao {
                                                            .setEffectiveDate(new DateTime(DateTimeZone.UTC))
                                                            .build();
 
-        dao.savePaymentInfo(paymentInfo);
+        paymentDao.savePaymentInfo(paymentInfo);
 
-        dao.updatePaymentInfo("CreditCard", paymentInfo.getPaymentId(), "Visa", "US");
+        paymentDao.updatePaymentInfo("CreditCard", paymentInfo.getPaymentId(), "Visa", "US");
 
     }
 
+    @Test
+    public void testGetPaymentForInvoice() throws AccountApiException {
+        final UUID invoiceId = UUID.randomUUID();
+        final UUID paymentAttemptId = UUID.randomUUID();
+        final UUID accountId = UUID.randomUUID();
+        final String paymentId = UUID.randomUUID().toString();
+        final BigDecimal invoiceAmount = BigDecimal.TEN;
+
+        final DateTime now = new DateTime(DateTimeZone.UTC);
+
+        PaymentAttempt originalPaymenAttempt = new PaymentAttempt(paymentAttemptId, invoiceId, accountId, invoiceAmount, Currency.USD, now, now, paymentId, null, null);
+
+        PaymentAttempt attempt = paymentDao.createPaymentAttempt(originalPaymenAttempt);
+
+        PaymentAttempt attempt2 = paymentDao.getPaymentAttemptForInvoiceId(invoiceId.toString());
+
+        Assert.assertEquals(attempt, attempt2);
+
+        PaymentAttempt attempt3 = paymentDao.getPaymentAttemptsForInvoiceIds(Arrays.asList(invoiceId.toString())).get(0);
+
+        Assert.assertEquals(attempt, attempt3);
+
+        PaymentAttempt attempt4 = paymentDao.getPaymentAttemptById(attempt3.getPaymentAttemptId());
+
+        Assert.assertEquals(attempt3, attempt4);
+
+        PaymentInfo originalPaymentInfo = new PaymentInfo.Builder().setPaymentId(paymentId)
+                                                           .setAmount(invoiceAmount)
+                                                           .setStatus("Processed")
+                                                           .setBankIdentificationNumber("1234")
+                                                           .setPaymentNumber("12345")
+                                                           .setPaymentMethodId("12345")
+                                                           .setReferenceId("12345")
+                                                           .setType("Electronic")
+                                                           .setCreatedDate(now)
+                                                           .setUpdatedDate(now)
+                                                           .setEffectiveDate(now)
+                                                           .build();
+
+        paymentDao.savePaymentInfo(originalPaymentInfo);
+        PaymentInfo paymentInfo = paymentDao.getPaymentInfo(Arrays.asList(invoiceId.toString())).get(0);
+
+        Assert.assertEquals(originalPaymentInfo, paymentInfo);
+    }
+
 }

diff --git a/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDaoWithEmbeddedDb.java b/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDaoWithEmbeddedDb.java
index da48c03..19ca39d 100644
--- a/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDaoWithEmbeddedDb.java
+++ b/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDaoWithEmbeddedDb.java
@@ -26,28 +26,25 @@ import org.testng.annotations.Test;
 
 import com.ning.billing.dbi.MysqlTestingHelper;
 
-public class TestPaymentDaoWithEmbeddedDb
-{
-    @Test(enabled = true, groups = { "slow", "database" })
-    public class TestPaymentDaoWithEmbeddedDB extends TestPaymentDao {
-        private final MysqlTestingHelper helper = new MysqlTestingHelper();
-
-        @BeforeClass(alwaysRun = true)
-        public void startMysql() throws IOException {
-            final String paymentddl = IOUtils.toString(MysqlTestingHelper.class.getResourceAsStream("/com/ning/billing/payment/ddl.sql"));
-
-            helper.startMysql();
-            helper.initDb(paymentddl);
-        }
-
-        @AfterClass(alwaysRun = true)
-        public void stopMysql() {
-            helper.stopMysql();
-        }
-
-        @BeforeMethod(alwaysRun = true)
-        public void setUp() throws IOException {
-            dao = new DefaultPaymentDao(helper.getDBI());
-        }
+@Test(enabled = true, groups = { "slow", "database" })
+public class TestPaymentDaoWithEmbeddedDb extends TestPaymentDao {
+    private final MysqlTestingHelper helper = new MysqlTestingHelper();
+
+    @BeforeClass(alwaysRun = true)
+    public void startMysql() throws IOException {
+        final String paymentddl = IOUtils.toString(MysqlTestingHelper.class.getResourceAsStream("/com/ning/billing/payment/ddl.sql"));
+
+        helper.startMysql();
+        helper.initDb(paymentddl);
+    }
+
+    @AfterClass(alwaysRun = true)
+    public void stopMysql() {
+        helper.stopMysql();
+    }
+
+    @BeforeMethod(alwaysRun = true)
+    public void setUp() throws IOException {
+        paymentDao = new DefaultPaymentDao(helper.getDBI());
     }
 }

diff --git a/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDaoWithMock.java b/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDaoWithMock.java
index f5af240..6e31f90 100644
--- a/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDaoWithMock.java
+++ b/payment/src/test/java/com/ning/billing/payment/dao/TestPaymentDaoWithMock.java
@@ -25,6 +25,6 @@ import org.testng.annotations.Test;
 public class TestPaymentDaoWithMock extends TestPaymentDao {
     @BeforeMethod(alwaysRun = true)
     public void setUp() throws IOException {
-        dao = new MockPaymentDao();
+        paymentDao = new MockPaymentDao();
     }
 }
\ No newline at end of file

diff --git a/payment/src/test/java/com/ning/billing/payment/TestNotifyInvoicePaymentApi.java b/payment/src/test/java/com/ning/billing/payment/TestNotifyInvoicePaymentApi.java
index 80de67f..e691f25 100644
--- a/payment/src/test/java/com/ning/billing/payment/TestNotifyInvoicePaymentApi.java
+++ b/payment/src/test/java/com/ning/billing/payment/TestNotifyInvoicePaymentApi.java
@@ -20,7 +20,6 @@ import static org.testng.Assert.assertNotNull;
 
 import java.util.UUID;
 
-import com.ning.billing.invoice.api.InvoicePayment;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Guice;
@@ -31,8 +30,10 @@ import com.ning.billing.account.api.Account;
 import com.ning.billing.account.api.AccountApiException;
 import com.ning.billing.account.glue.AccountModuleWithMocks;
 import com.ning.billing.invoice.api.Invoice;
+import com.ning.billing.invoice.api.InvoicePayment;
 import com.ning.billing.invoice.api.InvoicePaymentApi;
 import com.ning.billing.invoice.glue.InvoiceModuleWithMocks;
+import com.ning.billing.payment.api.PaymentAttempt;
 import com.ning.billing.payment.setup.PaymentTestModuleWithMocks;
 import com.ning.billing.util.bus.Bus;
 import com.ning.billing.util.bus.Bus.EventBusException;

diff --git a/payment/src/test/java/com/ning/billing/payment/TestPaymentProvider.java b/payment/src/test/java/com/ning/billing/payment/TestPaymentProvider.java
index 2c0aa13..99870b7 100644
--- a/payment/src/test/java/com/ning/billing/payment/TestPaymentProvider.java
+++ b/payment/src/test/java/com/ning/billing/payment/TestPaymentProvider.java
@@ -18,7 +18,6 @@ package com.ning.billing.payment;
 
 import static com.jayway.awaitility.Awaitility.await;
 import static java.util.concurrent.TimeUnit.MINUTES;
-import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
@@ -90,23 +89,5 @@ public class TestPaymentProvider {
         assertFalse(paymentInfoReceiver.getProcessedPayments().isEmpty());
         assertTrue(paymentInfoReceiver.getErrors().isEmpty());
 
-        final PaymentInfo paymentInfo = paymentInfoReceiver.getProcessedPayments().get(0);
-        final PaymentInfoRequest paymentInfoRequest = new PaymentInfoRequest(account.getId(), paymentInfo.getPaymentId());
-
-        paymentInfoReceiver.clear();
-        eventBus.post(paymentInfoRequest);
-        await().atMost(5, MINUTES).until(new Callable<Boolean>() {
-            @Override
-            public Boolean call() throws Exception {
-                List<PaymentInfo> processedPayments = paymentInfoReceiver.getProcessedPayments();
-                List<PaymentError> errors = paymentInfoReceiver.getErrors();
-
-                return processedPayments.size() == 1 || errors.size() == 1;
-            }
-        });
-
-        assertFalse(paymentInfoReceiver.getProcessedPayments().isEmpty());
-        assertTrue(paymentInfoReceiver.getErrors().isEmpty());
-        assertEquals(paymentInfoReceiver.getProcessedPayments().get(0), paymentInfo);
     }
 }

diff --git a/pom.xml b/pom.xml
index 179d4c3..c668551 100644
--- a/pom.xml
+++ b/pom.xml
@@ -436,7 +436,7 @@
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-surefire-plugin</artifactId>
-                <version>2.6</version>
+                <version>2.11</version>
                 <configuration>
                     <useManifestOnlyJar>false</useManifestOnlyJar>
                     <systemPropertyVariables>

diff --git a/util/src/main/java/com/ning/billing/util/glue/TagStoreModule.java b/util/src/main/java/com/ning/billing/util/glue/TagStoreModule.java
index 10651be..a598275 100644
--- a/util/src/main/java/com/ning/billing/util/glue/TagStoreModule.java
+++ b/util/src/main/java/com/ning/billing/util/glue/TagStoreModule.java
@@ -18,8 +18,6 @@ package com.ning.billing.util.glue;
 
 import com.google.inject.AbstractModule;
 import com.ning.billing.util.api.TagDefinitionUserApi;
-import com.ning.billing.util.clock.Clock;
-import com.ning.billing.util.clock.DefaultClock;
 import com.ning.billing.util.tag.api.DefaultTagDefinitionUserApi;
 import com.ning.billing.util.tag.dao.DefaultTagDefinitionDao;
 import com.ning.billing.util.tag.dao.TagDefinitionDao;
@@ -28,13 +26,16 @@ import com.ning.billing.util.tag.dao.TagStoreSqlDao;
 
 public class TagStoreModule extends AbstractModule
 {
-    @Override
-    protected void configure()
-    {
+    protected void installDaos() {
         bind(TagDefinitionSqlDao.class).toProvider(TagDescriptionDaoProvider.class).asEagerSingleton();
         bind(TagDefinitionDao.class).to(DefaultTagDefinitionDao.class).asEagerSingleton();
         bind(TagStoreSqlDao.class).toProvider(TagStoreDaoProvider.class).asEagerSingleton();
+    }
+
+    @Override
+    protected void configure()
+    {
+        installDaos();
         bind(TagDefinitionUserApi.class).to(DefaultTagDefinitionUserApi.class).asEagerSingleton();
     }
-    
 }

diff --git a/util/src/test/java/com/ning/billing/util/tag/dao/MockTagDefinitionDao.java b/util/src/test/java/com/ning/billing/util/tag/dao/MockTagDefinitionDao.java
new file mode 100644
index 0000000..f09e851
--- /dev/null
+++ b/util/src/test/java/com/ning/billing/util/tag/dao/MockTagDefinitionDao.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2010-2011 Ning, Inc.
+ *
+ * Ning 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 com.ning.billing.util.tag.dao;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.joda.time.DateTime;
+
+import com.ning.billing.util.api.TagDefinitionApiException;
+import com.ning.billing.util.tag.DefaultTagDefinition;
+import com.ning.billing.util.tag.TagDefinition;
+
+public class MockTagDefinitionDao implements TagDefinitionDao {
+    private final Map<String, TagDefinition> tags = new ConcurrentHashMap<String, TagDefinition>();
+
+    @Override
+    public List<TagDefinition> getTagDefinitions() {
+        return new ArrayList<TagDefinition>(tags.values());
+    }
+
+    @Override
+    public TagDefinition getByName(String definitionName) {
+        return tags.get(definitionName);
+    }
+
+    @Override
+    public TagDefinition create(String definitionName, String description, String createdBy) throws TagDefinitionApiException {
+        TagDefinition tag = new DefaultTagDefinition(UUID.randomUUID(), definitionName, description, createdBy, new DateTime());
+
+        tags.put(definitionName, tag);
+        return tag;
+    }
+
+    @Override
+    public void deleteAllTagsForDefinition(String definitionName) throws TagDefinitionApiException {
+        tags.remove(definitionName);
+    }
+
+    @Override
+    public void deleteTagDefinition(String definitionName) throws TagDefinitionApiException {
+        tags.remove(definitionName);
+    }
+}