killbill-memoizeit

diff --git a/entitlement/src/main/java/org/killbill/billing/entitlement/api/DefaultEntitlementApi.java b/entitlement/src/main/java/org/killbill/billing/entitlement/api/DefaultEntitlementApi.java
index 88f580a..8df759d 100644
--- a/entitlement/src/main/java/org/killbill/billing/entitlement/api/DefaultEntitlementApi.java
+++ b/entitlement/src/main/java/org/killbill/billing/entitlement/api/DefaultEntitlementApi.java
@@ -222,6 +222,13 @@ public class DefaultEntitlementApi extends DefaultEntitlementApiBase implements 
                 final InternalCallContext contextWithValidAccountRecordId = internalCallContextFactory.createInternalCallContext(accountId, callContext);
 
                 try {
+                    // First verify bundleKey
+                    for (final BaseEntitlementWithAddOnsSpecifier baseEntitlementWithAddOnsSpecifier : baseEntitlementWithAddOnsSpecifiers) {
+                        if (entitlementUtils.getFirstActiveSubscriptionIdForKeyOrNull(baseEntitlementWithAddOnsSpecifier.getExternalKey(), contextWithValidAccountRecordId) != null) {
+                            throw new EntitlementApiException(new SubscriptionBaseApiException(ErrorCode.SUB_CREATE_ACTIVE_BUNDLE_KEY_EXISTS, baseEntitlementWithAddOnsSpecifier.getExternalKey()));
+                        }
+                    }
+
                     final List<SubscriptionBaseWithAddOns> subscriptionsWithAddOns = subscriptionBaseInternalApi.createBaseSubscriptionsWithAddOns(accountId, baseEntitlementWithAddOnsSpecifiers, contextWithValidAccountRecordId);
                     final Map<BlockingState, UUID> blockingStateMap = new HashMap<BlockingState, UUID>();
                     int i = 0;

diff --git a/subscription/src/main/java/org/killbill/billing/subscription/api/svcs/DefaultSubscriptionInternalApi.java b/subscription/src/main/java/org/killbill/billing/subscription/api/svcs/DefaultSubscriptionInternalApi.java
index b190af1..c72cf42 100644
--- a/subscription/src/main/java/org/killbill/billing/subscription/api/svcs/DefaultSubscriptionInternalApi.java
+++ b/subscription/src/main/java/org/killbill/billing/subscription/api/svcs/DefaultSubscriptionInternalApi.java
@@ -54,6 +54,7 @@ import org.killbill.billing.entitlement.api.Entitlement.EntitlementState;
 import org.killbill.billing.entitlement.api.EntitlementAOStatusDryRun;
 import org.killbill.billing.entitlement.api.EntitlementAOStatusDryRun.DryRunChangeReason;
 import org.killbill.billing.entitlement.api.EntitlementSpecifier;
+import org.killbill.billing.entitlement.api.Subscription;
 import org.killbill.billing.events.EffectiveSubscriptionInternalEvent;
 import org.killbill.billing.invoice.api.DryRunArguments;
 import org.killbill.billing.subscription.api.SubscriptionApiBase;
@@ -339,6 +340,33 @@ public class DefaultSubscriptionInternalApi extends SubscriptionApiBase implemen
     public SubscriptionBaseBundle createBundleForAccount(final UUID accountId, final String bundleKey, final InternalCallContext context) throws SubscriptionBaseApiException {
 
         final List<SubscriptionBaseBundle> existingBundles = dao.getSubscriptionBundlesForKey(bundleKey, context);
+
+        //
+        // Because the creation of the SubscriptionBundle is not atomic (with creation of Subscription/SubscriptionEvent), we verify if we were left
+        // with an empty SubscriptionBaseBundle form a past failing operation (See #684). We only allow reuse if such SubscriptionBaseBundle is fully
+        // empty (and don't allow use case where all Subscription are cancelled, which is the condition for that key to be re-used)
+        // Such condition should have been checked upstream (to decide whether that key is valid or not)
+        //
+        final SubscriptionBaseBundle existingBundleForAccount = Iterables.tryFind(existingBundles, new Predicate<SubscriptionBaseBundle>() {
+            @Override
+            public boolean apply(final SubscriptionBaseBundle input) {
+                return input.getAccountId().equals(accountId);
+            }
+        }).orNull();
+
+        // If Bundle already exists, and there is 0 Subscription, we reuse
+        if (existingBundleForAccount != null) {
+            try {
+                final Map<UUID, List<SubscriptionBase>> accountSubscriptions = dao.getSubscriptionsForAccount(context);
+                final List<SubscriptionBase> subscriptions = accountSubscriptions.get(existingBundleForAccount.getId());
+                if (subscriptions == null || subscriptions.size() == 0) {
+                    return existingBundleForAccount;
+                }
+            } catch (final CatalogApiException e) {
+                throw new SubscriptionBaseApiException(e);
+            }
+        }
+
         final DateTime now = clock.getUTCNow();
         final DateTime originalCreatedDate = !existingBundles.isEmpty() ? existingBundles.get(0).getCreatedDate() : now;
         final DefaultSubscriptionBaseBundle bundle = new DefaultSubscriptionBaseBundle(bundleKey, accountId, now, originalCreatedDate, now, now);