killbill-memoizeit

diff --git a/payment/src/main/java/org/killbill/billing/payment/core/PaymentProcessor.java b/payment/src/main/java/org/killbill/billing/payment/core/PaymentProcessor.java
index 0af54d5..53e1606 100644
--- a/payment/src/main/java/org/killbill/billing/payment/core/PaymentProcessor.java
+++ b/payment/src/main/java/org/killbill/billing/payment/core/PaymentProcessor.java
@@ -322,7 +322,6 @@ public class PaymentProcessor extends ProcessorBase {
                                      final boolean shouldLockAccountAndDispatch, @Nullable final OperationResult overridePluginOperationResult,
                                      final Iterable<PluginProperty> properties,
                                      final CallContext callContext, final InternalCallContext internalCallContext) throws PaymentApiException {
-        validateUniqueTransactionExternalKey(paymentTransactionExternalKey, internalCallContext);
         final UUID nonNullPaymentId = paymentAutomatonRunner.run(isApiPayment,
                                                                  transactionType,
                                                                  account,

diff --git a/payment/src/main/java/org/killbill/billing/payment/core/ProcessorBase.java b/payment/src/main/java/org/killbill/billing/payment/core/ProcessorBase.java
index ae4e3a2..f8c2d67 100644
--- a/payment/src/main/java/org/killbill/billing/payment/core/ProcessorBase.java
+++ b/payment/src/main/java/org/killbill/billing/payment/core/ProcessorBase.java
@@ -164,27 +164,8 @@ public abstract class ProcessorBase {
         return internalCallContextFactory.createCallContext(context);
     }
 
-    protected void validateUniqueTransactionExternalKey(@Nullable final String transactionExternalKey, final InternalTenantContext tenantContext) throws PaymentApiException {
-        // If no key specified, system will allocate a unique one later.
-        if (transactionExternalKey == null) {
-            return;
-        }
-
-        final List<PaymentTransactionModelDao> transactions = paymentDao.getPaymentTransactionsByExternalKey(transactionExternalKey, tenantContext);
-        final PaymentTransactionModelDao transactionAlreadyExists = Iterables.tryFind(transactions, new Predicate<PaymentTransactionModelDao>() {
-            @Override
-            public boolean apply(final PaymentTransactionModelDao input) {
-                return input.getTransactionStatus() == TransactionStatus.SUCCESS;
-            }
-        }).orNull();
-        if (transactionAlreadyExists != null) {
-            throw new PaymentApiException(ErrorCode.PAYMENT_ACTIVE_TRANSACTION_KEY_EXISTS, transactionExternalKey);
-        }
-    }
-
     // TODO Rename - there is no lock!
     public interface WithAccountLockCallback<PluginDispatcherReturnType, ExceptionType extends Exception> {
-
         public PluginDispatcherReturnType doOperation() throws ExceptionType;
     }
 

diff --git a/payment/src/main/java/org/killbill/billing/payment/core/sm/control/DefaultControlCompleted.java b/payment/src/main/java/org/killbill/billing/payment/core/sm/control/DefaultControlCompleted.java
index e36fa20..3e42fee 100644
--- a/payment/src/main/java/org/killbill/billing/payment/core/sm/control/DefaultControlCompleted.java
+++ b/payment/src/main/java/org/killbill/billing/payment/core/sm/control/DefaultControlCompleted.java
@@ -17,6 +17,7 @@
 
 package org.killbill.billing.payment.core.sm.control;
 
+import java.util.List;
 import java.util.UUID;
 
 import org.killbill.automaton.Operation.OperationCallback;
@@ -25,10 +26,15 @@ import org.killbill.automaton.State;
 import org.killbill.automaton.State.EnteringStateCallback;
 import org.killbill.automaton.State.LeavingStateCallback;
 import org.killbill.billing.ObjectType;
+import org.killbill.billing.payment.api.TransactionStatus;
 import org.killbill.billing.payment.core.sm.PluginRoutingPaymentAutomatonRunner;
 import org.killbill.billing.payment.dao.PaymentAttemptModelDao;
+import org.killbill.billing.payment.dao.PaymentTransactionModelDao;
 import org.killbill.billing.payment.retry.BaseRetryService.RetryServiceScheduler;
 
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+
 public class DefaultControlCompleted implements EnteringStateCallback {
 
     private PluginRoutingPaymentAutomatonRunner retryablePaymentAutomatonRunner;
@@ -50,9 +56,31 @@ public class DefaultControlCompleted implements EnteringStateCallback {
                                    null;
         retryablePaymentAutomatonRunner.getPaymentDao().updatePaymentAttempt(attempt.getId(), transactionId, state.getName(), paymentStateContext.getInternalCallContext());
 
-        if ("RETRIED".equals(state.getName())) {
+        if ("RETRIED".equals(state.getName()) && !isUnknownTransaction()) {
             retryServiceScheduler.scheduleRetry(ObjectType.PAYMENT_ATTEMPT, attempt.getId(), attempt.getId(), attempt.getTenantRecordId(),
                                                 paymentStateContext.getPaymentControlPluginNames(), paymentStateContext.getRetryDate());
         }
     }
+
+    //
+    // If we see an UNKNOWN transaction we prevent it to be rescheduled as the Janitor will *try* to fix it, and that could lead to infinite retries from a badly behaved plugin
+    // (In other words, plugin should ONLY retry 'known' transaction)
+    //
+    private boolean isUnknownTransaction() {
+        if (paymentStateContext.getCurrentTransaction() != null) {
+            return paymentStateContext.getCurrentTransaction().getTransactionStatus() == TransactionStatus.UNKNOWN;
+        } else {
+            final List<PaymentTransactionModelDao> transactions = retryablePaymentAutomatonRunner.getPaymentDao().getPaymentTransactionsByExternalKey(paymentStateContext.getPaymentTransactionExternalKey(), paymentStateContext.getInternalCallContext());
+            return Iterables.any(transactions, new Predicate<PaymentTransactionModelDao>() {
+                @Override
+                public boolean apply(final PaymentTransactionModelDao input) {
+                    return input.getTransactionStatus() == TransactionStatus.UNKNOWN &&
+                           // Not strictly required
+                           // (Note, we don't match on AttemptId as it is risky, the row on disk would match the first attempt, not necessarily the current one)
+                           input.getAccountRecordId() == paymentStateContext.getInternalCallContext().getAccountRecordId() &&
+                           input.getAmount() == paymentStateContext.getAmount();
+                }
+            });
+        }
+    }
 }

diff --git a/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentEnteringStateCallback.java b/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentEnteringStateCallback.java
index 8cdc6eb..9860f6b 100644
--- a/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentEnteringStateCallback.java
+++ b/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentEnteringStateCallback.java
@@ -52,6 +52,10 @@ public abstract class PaymentEnteringStateCallback implements EnteringStateCallb
     public void enteringState(final State newState, final Operation.OperationCallback operationCallback, final OperationResult operationResult, final LeavingStateCallback leavingStateCallback) {
         logger.debug("Entering state {} with result {}", newState.getName(), operationResult);
 
+        if (paymentStateContext.isSkipOperationForUnknownTransaction()) {
+            return;
+        }
+
         // If the transaction was not created -- for instance we had an exception in leavingState callback then we bail; if not, then update state:
         if (paymentStateContext.getPaymentTransactionModelDao() != null && paymentStateContext.getPaymentTransactionModelDao().getId() != null) {
             final PaymentTransactionInfoPlugin paymentInfoPlugin = paymentStateContext.getPaymentTransactionInfoPlugin();

diff --git a/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentLeavingStateCallback.java b/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentLeavingStateCallback.java
index 18dd659..d898745 100644
--- a/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentLeavingStateCallback.java
+++ b/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentLeavingStateCallback.java
@@ -17,17 +17,24 @@
 
 package org.killbill.billing.payment.core.sm.payments;
 
+import java.util.List;
+
 import org.killbill.automaton.OperationException;
 import org.killbill.automaton.State;
 import org.killbill.automaton.State.LeavingStateCallback;
 import org.killbill.billing.ErrorCode;
 import org.killbill.billing.payment.api.PaymentApiException;
+import org.killbill.billing.payment.api.TransactionStatus;
 import org.killbill.billing.payment.core.sm.PaymentAutomatonDAOHelper;
 import org.killbill.billing.payment.core.sm.PaymentStateContext;
 import org.killbill.billing.payment.dao.PaymentTransactionModelDao;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.base.Predicate;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+
 public abstract class PaymentLeavingStateCallback implements LeavingStateCallback {
 
     private final Logger logger = LoggerFactory.getLogger(PaymentLeavingStateCallback.class);
@@ -51,16 +58,84 @@ public abstract class PaymentLeavingStateCallback implements LeavingStateCallbac
                 throw new PaymentApiException(ErrorCode.PAYMENT_NO_DEFAULT_PAYMENT_METHOD, paymentStateContext.getAccount().getId());
             }
 
-            // If the transactionId has been specified, it means this is an operation in several stage (INIT -> PENDING -> XXX), so we don't need to create the row,
-            // but we do need to set the transactionModelDao in the context so enteringState logic can take place.
-            if (paymentStateContext.getTransactionId() == null) {
-                daoHelper.createNewPaymentTransaction();
-            } else {
+            //
+            // Extract existing transaction matching the transactionId if specified (for e.g notifyPendingTransactionOfStateChanged), or based on transactionExternalKey
+            //
+            final List<PaymentTransactionModelDao> existingPaymentTransactions;
+            if (paymentStateContext.getTransactionId() != null) {
                 final PaymentTransactionModelDao transactionModelDao = daoHelper.getPaymentDao().getPaymentTransaction(paymentStateContext.getTransactionId(), paymentStateContext.getInternalCallContext());
-                paymentStateContext.setPaymentTransactionModelDao(transactionModelDao);
+                existingPaymentTransactions = ImmutableList.of(transactionModelDao);
+            } else {
+                existingPaymentTransactions = daoHelper.getPaymentDao().getPaymentTransactionsByExternalKey(paymentStateContext.getPaymentTransactionExternalKey(), paymentStateContext.getInternalCallContext());
+            }
+
+            // Validate some constraints on the unicity of that paymentTransactionExternalKey
+            validateUniqueTransactionExternalKey(existingPaymentTransactions);
+
+            // Handle UNKNOWN cases, where we skip the whole state machine and let the getPayment logic refresh the state.
+            final PaymentTransactionModelDao unknownPaymentTransaction = getUnknownPaymentTransaction(existingPaymentTransactions);
+            if (unknownPaymentTransaction != null) {
+                // Reset the attemptId on the existing paymentTransaction row since it it is not accurate
+                unknownPaymentTransaction.setAttemptId(paymentStateContext.getAttemptId());
+                // Set the current paymentTransaction in the context (needed for the state machine logic)
+                paymentStateContext.setPaymentTransactionModelDao(unknownPaymentTransaction);
+                // Set special flag to bypass the state machine altogether (plugin will not be called, state will not be updated, no event will be sent)
+                paymentStateContext.setSkipOperationForUnknownTransaction(true);
+                return;
+            }
+
+            // Handle PENDING cases, where we want to re-use the same transaction
+            final PaymentTransactionModelDao pendingPaymentTransaction = getPendingPaymentTransaction(existingPaymentTransactions);
+            if (pendingPaymentTransaction != null) {
+                // Set the current paymentTransaction in the context (needed for the state machine logic)
+                paymentStateContext.setPaymentTransactionModelDao(pendingPaymentTransaction);
+                return;
             }
+
+            // At this point we are left with PAYMENT_FAILURE, PLUGIN_FAILURE or nothing, and we validated the uniquess of the paymentTransactionExternalKey so we will create a new row
+            daoHelper.createNewPaymentTransaction();
+
         } catch (PaymentApiException e) {
             throw new OperationException(e);
         }
     }
+
+    protected PaymentTransactionModelDao getUnknownPaymentTransaction(final List<PaymentTransactionModelDao> existingPaymentTransactions) throws PaymentApiException {
+        return Iterables.tryFind(existingPaymentTransactions, new Predicate<PaymentTransactionModelDao>() {
+            @Override
+            public boolean apply(final PaymentTransactionModelDao input) {
+                return input.getTransactionStatus() == TransactionStatus.UNKNOWN;
+            }
+        }).orNull();
+    }
+
+    protected PaymentTransactionModelDao getPendingPaymentTransaction(final List<PaymentTransactionModelDao> existingPaymentTransactions) throws PaymentApiException {
+        return Iterables.tryFind(existingPaymentTransactions, new Predicate<PaymentTransactionModelDao>() {
+            @Override
+            public boolean apply(final PaymentTransactionModelDao input) {
+                return input.getTransactionStatus() == TransactionStatus.PENDING;
+            }
+        }).orNull();
+    }
+
+    protected void validateUniqueTransactionExternalKey(final List<PaymentTransactionModelDao> existingPaymentTransactions) throws PaymentApiException {
+        // If no key specified, system will allocate a unique one later, there is nothing to check
+        if (paymentStateContext.getPaymentTransactionExternalKey() == null) {
+            return;
+        }
+
+        if (Iterables.any(existingPaymentTransactions, new Predicate<PaymentTransactionModelDao>() {
+            @Override
+            public boolean apply(final PaymentTransactionModelDao input) {
+                       // An existing transaction in a SUCCESS state
+                return input.getTransactionStatus() == TransactionStatus.SUCCESS ||
+                       // Or, an existing transaction for a different payment (to do really well, we should also check on paymentExternalKey which is not available here)
+                       (paymentStateContext.getPaymentId() != null && input.getPaymentId().compareTo(paymentStateContext.getPaymentId()) != 0) ||
+                       // Or, an existing transaction for a different account.
+                       (input.getAccountRecordId() != paymentStateContext.getInternalCallContext().getAccountRecordId());
+            }
+        })) {
+            throw new PaymentApiException(ErrorCode.PAYMENT_ACTIVE_TRANSACTION_KEY_EXISTS, paymentStateContext.getPaymentTransactionExternalKey());
+        }
+    }
 }

diff --git a/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentOperation.java b/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentOperation.java
index 6781ecd..6d57033 100644
--- a/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentOperation.java
+++ b/payment/src/main/java/org/killbill/billing/payment/core/sm/payments/PaymentOperation.java
@@ -66,6 +66,11 @@ public abstract class PaymentOperation extends OperationCallbackBase<PaymentTran
 
     @Override
     public OperationResult doOperationCallback() throws OperationException {
+
+        if (paymentStateContext.isSkipOperationForUnknownTransaction()) {
+            return OperationResult.SUCCESS;
+        }
+
         try {
             this.plugin = daoHelper.getPaymentProviderPlugin();
 

diff --git a/payment/src/main/java/org/killbill/billing/payment/core/sm/PaymentStateContext.java b/payment/src/main/java/org/killbill/billing/payment/core/sm/PaymentStateContext.java
index 034fbc4..d9c097c 100644
--- a/payment/src/main/java/org/killbill/billing/payment/core/sm/PaymentStateContext.java
+++ b/payment/src/main/java/org/killbill/billing/payment/core/sm/PaymentStateContext.java
@@ -50,6 +50,7 @@ public class PaymentStateContext {
     protected String paymentTransactionExternalKey;
     protected Currency currency;
     protected Iterable<PluginProperty> properties;
+    protected boolean skipOperationForUnknownTransaction;
 
     // Can be updated later via paymentTransactionModelDao (e.g. for auth or purchase)
     protected final UUID paymentId;
@@ -94,6 +95,7 @@ public class PaymentStateContext {
         this.internalCallContext = internalCallContext;
         this.callContext = callContext;
         this.onLeavingStateExistingTransactions = ImmutableList.of();
+        this.skipOperationForUnknownTransaction = false;
     }
 
     public boolean isApiPayment() {
@@ -199,4 +201,12 @@ public class PaymentStateContext {
     public CallContext getCallContext() {
         return callContext;
     }
+
+    public boolean isSkipOperationForUnknownTransaction() {
+        return skipOperationForUnknownTransaction;
+    }
+
+    public void setSkipOperationForUnknownTransaction(final boolean skipOperationForUnknownTransaction) {
+        this.skipOperationForUnknownTransaction = skipOperationForUnknownTransaction;
+    }
 }

diff --git a/payment/src/main/java/org/killbill/billing/payment/core/sm/PaymentStateMachineHelper.java b/payment/src/main/java/org/killbill/billing/payment/core/sm/PaymentStateMachineHelper.java
index dfea014..d6a12c4 100644
--- a/payment/src/main/java/org/killbill/billing/payment/core/sm/PaymentStateMachineHelper.java
+++ b/payment/src/main/java/org/killbill/billing/payment/core/sm/PaymentStateMachineHelper.java
@@ -57,7 +57,7 @@ public class PaymentStateMachineHelper {
     private static final String VOID_SUCCESS = "VOID_SUCCESS";
     private static final String CHARGEBACK_SUCCESS = "CHARGEBACK_SUCCESS";
 
-    private static final String AUTHORIZE_PENDING = "AUTHORIZE_PENDING";
+    private static final String AUTHORIZE_PENDING = "AUTH_PENDING";
 
     private static final String AUTHORIZE_FAILED = "AUTH_FAILED";
     private static final String CAPTURE_FAILED = "CAPTURE_FAILED";

diff --git a/payment/src/main/java/org/killbill/billing/payment/invoice/InvoicePaymentRoutingPluginApi.java b/payment/src/main/java/org/killbill/billing/payment/invoice/InvoicePaymentRoutingPluginApi.java
index 30d5115..e97043d 100644
--- a/payment/src/main/java/org/killbill/billing/payment/invoice/InvoicePaymentRoutingPluginApi.java
+++ b/payment/src/main/java/org/killbill/billing/payment/invoice/InvoicePaymentRoutingPluginApi.java
@@ -354,10 +354,10 @@ public final class InvoicePaymentRoutingPluginApi implements PaymentRoutingPlugi
             case PAYMENT_FAILURE:
                 return getNextRetryDateForPaymentFailure(purchasedTransactions);
 
-            case UNKNOWN:
             case PLUGIN_FAILURE:
                 return getNextRetryDateForPluginFailure(purchasedTransactions);
 
+            case UNKNOWN:
             default:
                 return null;
         }
@@ -385,8 +385,7 @@ public final class InvoicePaymentRoutingPluginApi implements PaymentRoutingPlugi
     private DateTime getNextRetryDateForPluginFailure(final List<PaymentTransactionModelDao> purchasedTransactions) {
 
         DateTime result = null;
-        // Not clear whether or not we should really include TransactionStatus.UNKNOWN
-        final int attemptsInState = getNumberAttemptsInState(purchasedTransactions, TransactionStatus.PLUGIN_FAILURE, TransactionStatus.UNKNOWN);
+        final int attemptsInState = getNumberAttemptsInState(purchasedTransactions, TransactionStatus.PLUGIN_FAILURE);
         final int retryAttempt = (attemptsInState - 1) >= 0 ? (attemptsInState - 1) : 0;
 
         if (retryAttempt < paymentConfig.getPluginFailureRetryMaxAttempts()) {

diff --git a/payment/src/test/java/org/killbill/billing/payment/api/TestPaymentApi.java b/payment/src/test/java/org/killbill/billing/payment/api/TestPaymentApi.java
index e5bb1ee..9732b7c 100644
--- a/payment/src/test/java/org/killbill/billing/payment/api/TestPaymentApi.java
+++ b/payment/src/test/java/org/killbill/billing/payment/api/TestPaymentApi.java
@@ -440,7 +440,7 @@ public class TestPaymentApi extends PaymentTestSuiteWithEmbeddedDB {
         final String transactionExternalKey2 = "couyc";
 
         final Payment payment = paymentApi.createPurchase(account, account.getPaymentMethodId(), null, requestedAmount, Currency.AED, paymentExternalKey, transactionExternalKey,
-                                                                ImmutableList.<PluginProperty>of(), callContext);
+                                                          ImmutableList.<PluginProperty>of(), callContext);
 
         paymentApi.createChargeback(account, payment.getId(), requestedAmount, Currency.AED, transactionExternalKey2,  callContext);
         final Payment payment2 = paymentApi.getPayment(payment.getId(), false, ImmutableList.<PluginProperty>of(), callContext);
@@ -527,7 +527,7 @@ public class TestPaymentApi extends PaymentTestSuiteWithEmbeddedDB {
         final BigDecimal requestedAmount = new BigDecimal("80.0091");
 
         final Payment initialPayment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(),
-                                                                            UUID.randomUUID().toString(), UUID.randomUUID().toString(), ImmutableList.<PluginProperty>of(), callContext);
+                                                                      UUID.randomUUID().toString(), UUID.randomUUID().toString(), ImmutableList.<PluginProperty>of(), callContext);
 
 
         try {
@@ -587,6 +587,71 @@ public class TestPaymentApi extends PaymentTestSuiteWithEmbeddedDB {
     }
 
 
+    @Test(groups = "slow")
+    public void testApiRetryWithUnknownPaymentTransaction() throws Exception {
+        final BigDecimal requestedAmount = BigDecimal.TEN;
+
+        final String paymentExternalKey = UUID.randomUUID().toString();
+        final String paymentTransactionExternalKey = UUID.randomUUID().toString();
+
+        final Payment badPayment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(),
+                                                                  paymentExternalKey, paymentTransactionExternalKey, ImmutableList.<PluginProperty>of(), callContext);
+
+
+        final String paymentStateName = paymentSMHelper.getErroredStateForTransaction(TransactionType.AUTHORIZE).toString();
+        paymentDao.updatePaymentAndTransactionOnCompletion(account.getId(), badPayment.getId(), TransactionType.AUTHORIZE, paymentStateName, paymentStateName,
+                                                           badPayment.getTransactions().get(0).getId(), TransactionStatus.UNKNOWN, requestedAmount, account.getCurrency(),
+                                                           "eroor 64", "bad something happened", internalCallContext);
+
+
+        final Payment payment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(),
+                                                               paymentExternalKey, paymentTransactionExternalKey, ImmutableList.<PluginProperty>of(), callContext);
+
+
+        Assert.assertEquals(payment.getId(), badPayment.getId());
+        Assert.assertEquals(payment.getExternalKey(), paymentExternalKey);
+        Assert.assertEquals(payment.getExternalKey(), paymentExternalKey);
+
+        Assert.assertEquals(payment.getTransactions().size(), 1);
+        Assert.assertEquals(payment.getTransactions().get(0).getTransactionStatus(), TransactionStatus.SUCCESS);
+        Assert.assertEquals(payment.getTransactions().get(0).getExternalKey(), paymentTransactionExternalKey);
+
+    }
+
+
+    // Example of a 3D secure payment for instnace
+    @Test(groups = "slow")
+    public void testApiWithPendingPaymentTransaction() throws Exception {
+        final BigDecimal requestedAmount = BigDecimal.TEN;
+
+        final String paymentExternalKey = UUID.randomUUID().toString();
+        final String paymentTransactionExternalKey = UUID.randomUUID().toString();
+
+        final Payment pendingPayment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), null, requestedAmount, account.getCurrency(),
+                                                                  paymentExternalKey, paymentTransactionExternalKey, ImmutableList.<PluginProperty>of(), callContext);
+
+
+        final String paymentStateName = paymentSMHelper.getPendingStateForTransaction(TransactionType.AUTHORIZE).toString();
+        paymentDao.updatePaymentAndTransactionOnCompletion(account.getId(), pendingPayment.getId(), TransactionType.AUTHORIZE, paymentStateName, paymentStateName,
+                                                           pendingPayment.getTransactions().get(0).getId(), TransactionStatus.PENDING, requestedAmount, account.getCurrency(),
+                                                           null, null, internalCallContext);
+
+
+        final Payment payment = paymentApi.createAuthorization(account, account.getPaymentMethodId(), pendingPayment.getId(), requestedAmount, account.getCurrency(),
+                                                               paymentExternalKey, paymentTransactionExternalKey, ImmutableList.<PluginProperty>of(), callContext);
+
+
+        Assert.assertEquals(payment.getId(), pendingPayment.getId());
+        Assert.assertEquals(payment.getExternalKey(), paymentExternalKey);
+        Assert.assertEquals(payment.getExternalKey(), paymentExternalKey);
+
+        Assert.assertEquals(payment.getTransactions().size(), 1);
+        Assert.assertEquals(payment.getTransactions().get(0).getTransactionStatus(), TransactionStatus.SUCCESS);
+        Assert.assertEquals(payment.getTransactions().get(0).getExternalKey(), paymentTransactionExternalKey);
+
+    }
+
+
 
 
     private List<PluginProperty> createPropertiesForInvoice(final Invoice invoice) {

diff --git a/payment/src/test/java/org/killbill/billing/payment/core/sm/TestPaymentLeavingStateCallback.java b/payment/src/test/java/org/killbill/billing/payment/core/sm/TestPaymentLeavingStateCallback.java
index 0097eaf..c063f80 100644
--- a/payment/src/test/java/org/killbill/billing/payment/core/sm/TestPaymentLeavingStateCallback.java
+++ b/payment/src/test/java/org/killbill/billing/payment/core/sm/TestPaymentLeavingStateCallback.java
@@ -22,8 +22,10 @@ import java.util.UUID;
 
 import javax.annotation.Nullable;
 
+import org.killbill.automaton.OperationException;
 import org.killbill.automaton.State;
 import org.killbill.billing.account.api.Account;
+import org.killbill.billing.callcontext.InternalCallContext;
 import org.killbill.billing.catalog.api.Currency;
 import org.killbill.billing.payment.PaymentTestSuiteWithEmbeddedDB;
 import org.killbill.billing.payment.api.PaymentApiException;
@@ -72,6 +74,8 @@ public class TestPaymentLeavingStateCallback extends PaymentTestSuiteWithEmbedde
         // Verify the payment has only one transaction
         Assert.assertEquals(paymentDao.getTransactionsForPayment(paymentId, internalCallContext).size(), 1);
 
+        // Reset paymentExternalKey to something else
+        paymentStateContext.setPaymentTransactionExternalKey(UUID.randomUUID().toString());
         callback.leavingState(state);
 
         // Verify a new transaction was created
@@ -81,6 +85,50 @@ public class TestPaymentLeavingStateCallback extends PaymentTestSuiteWithEmbedde
         Assert.assertEquals(paymentDao.getTransactionsForPayment(paymentId, internalCallContext).size(), 2);
     }
 
+    @Test(groups = "slow", expectedExceptions = OperationException.class)
+    public void testLeaveStateForConflictingPaymentTransactionExternalKey() throws Exception {
+        final UUID paymentId = UUID.randomUUID();
+        setUp(paymentId);
+
+        // Verify the payment has only one transaction
+        Assert.assertEquals(paymentDao.getTransactionsForPayment(paymentId, internalCallContext).size(), 1);
+
+        // Will validate the validateUniqueTransactionExternalKey logic for when we reuse the same payment transactionExternalKey
+        callback.leavingState(state);
+    }
+
+
+    @Test(groups = "slow", expectedExceptions = OperationException.class)
+    public void testLeaveStateForConflictingPaymentTransactionExternalKeyAcrossAccounts() throws Exception {
+        final UUID paymentId = UUID.randomUUID();
+        setUp(paymentId);
+
+        // Verify the payment has only one transaction
+        Assert.assertEquals(paymentDao.getTransactionsForPayment(paymentId, internalCallContext).size(), 1);
+
+        final InternalCallContext internalCallContextForOtherAccount =  new InternalCallContext(paymentStateContext.getInternalCallContext(), 123L);
+
+        paymentStateContext = new PaymentStateContext(true,
+                                                      paymentId,
+                                                      null,
+                                                      null,
+                                                      paymentStateContext.getPaymentExternalKey(),
+                                                      paymentStateContext.getPaymentTransactionExternalKey(),
+                                                      paymentStateContext.getTransactionType(),
+                                                      paymentStateContext.getAccount(),
+                                                      paymentStateContext.getPaymentMethodId(),
+                                                      paymentStateContext.getAmount(),
+                                                      paymentStateContext.getCurrency(),
+                                                      paymentStateContext.shouldLockAccountAndDispatch,
+                                                      paymentStateContext.overridePluginOperationResult,
+                                                      paymentStateContext.getProperties(),
+                                                      internalCallContextForOtherAccount,
+                                                      callContext);
+
+        callback.leavingState(state);
+    }
+
+
     private void verifyPaymentTransaction() {
         Assert.assertNotNull(paymentStateContext.getPaymentTransactionModelDao().getPaymentId());
         Assert.assertEquals(paymentStateContext.getPaymentTransactionModelDao().getTransactionExternalKey(), paymentStateContext.getPaymentTransactionExternalKey());
@@ -98,7 +146,8 @@ public class TestPaymentLeavingStateCallback extends PaymentTestSuiteWithEmbedde
         Mockito.when(account.getId()).thenReturn(UUID.randomUUID());
         paymentStateContext = new PaymentStateContext(true,
                                                       paymentId,
-                                                      null, null,
+                                                      null,
+                                                      null,
                                                       UUID.randomUUID().toString(),
                                                       UUID.randomUUID().toString(),
                                                       TransactionType.CAPTURE,

diff --git a/payment/src/test/java/org/killbill/billing/payment/TestRetryService.java b/payment/src/test/java/org/killbill/billing/payment/TestRetryService.java
index dcf9a9a..6a1d082 100644
--- a/payment/src/test/java/org/killbill/billing/payment/TestRetryService.java
+++ b/payment/src/test/java/org/killbill/billing/payment/TestRetryService.java
@@ -87,19 +87,11 @@ public class TestRetryService extends PaymentTestSuiteNoDB {
         return payment;
     }
 
-    @Test(groups = "fast")
-    public void testFailedPluginWithOneSuccessfulRetry() throws Exception {
-        testSchedulesRetryInternal(1, true, FailureType.PLUGIN_EXCEPTION);
-    }
-
-    @Test(groups = "fast")
-    public void testFailedPluginWithLastRetrySuccess() throws Exception {
-        testSchedulesRetryInternal(paymentConfig.getPluginFailureRetryMaxAttempts(), true, FailureType.PLUGIN_EXCEPTION);
-    }
 
+    // PLUGIN_EXCEPTION will lead to UNKNOWN row that will not be retried by the plugin
     @Test(groups = "fast")
     public void testAbortedPlugin() throws Exception {
-        testSchedulesRetryInternal(paymentConfig.getPluginFailureRetryMaxAttempts(), false, FailureType.PLUGIN_EXCEPTION);
+        testSchedulesRetryInternal(0, false, FailureType.PLUGIN_EXCEPTION);
     }
 
     @Test(groups = "fast")
@@ -242,7 +234,7 @@ public class TestRetryService extends PaymentTestSuiteNoDB {
         if (failureType == FailureType.PAYMENT_FAILURE) {
             return paymentConfig.getPaymentFailureRetryDays().size();
         } else {
-            return paymentConfig.getPluginFailureRetryMaxAttempts();
+            return 0;
         }
     }