killbill-aplcache

diff --git a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestWithPriceOverride.java b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestWithPriceOverride.java
index 5309ad9..638ea1d 100644
--- a/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestWithPriceOverride.java
+++ b/beatrix/src/test/java/org/killbill/billing/beatrix/integration/TestWithPriceOverride.java
@@ -27,7 +27,6 @@ import org.killbill.billing.account.api.AccountData;
 import org.killbill.billing.api.TestApiListener.NextEvent;
 import org.killbill.billing.beatrix.util.InvoiceChecker.ExpectedInvoiceItemCheck;
 import org.killbill.billing.catalog.DefaultPlanPhasePriceOverride;
-import org.killbill.billing.catalog.DefaultPriceListSet;
 import org.killbill.billing.catalog.api.BillingActionPolicy;
 import org.killbill.billing.catalog.api.BillingPeriod;
 import org.killbill.billing.catalog.api.PlanPhasePriceOverride;
@@ -59,7 +58,6 @@ public class TestWithPriceOverride extends TestIntegrationBase {
         invoiceChecker.checkInvoice(account.getId(), 1, callContext, new ExpectedInvoiceItemCheck(clock.getUTCToday(), null, InvoiceItemType.FIXED, new BigDecimal("1")));
     }
 
-
     @Test(groups = "slow")
     public void testCreateWithRecurringPriceOverride() throws Exception {
 
@@ -95,9 +93,6 @@ public class TestWithPriceOverride extends TestIntegrationBase {
         invoiceChecker.checkInvoice(account.getId(), 3, callContext, new ExpectedInvoiceItemCheck(new LocalDate(2012, 6, 1), new LocalDate(2012, 7, 1), InvoiceItemType.RECURRING, BigDecimal.TEN));
     }
 
-
-
-
     @Test(groups = "slow")
     public void testChangePlanWithRecurringPriceOverride() throws Exception {
 
@@ -109,7 +104,6 @@ public class TestWithPriceOverride extends TestIntegrationBase {
         // Set clock to the initial start date - we implicitly assume here that the account timezone is UTC
         clock.setDay(new LocalDate(2012, 4, 1));
 
-
         final DefaultEntitlement bpSubscription = createBaseEntitlementAndCheckForCompletion(account.getId(), "bundleKey", "Shotgun", ProductCategory.BASE, BillingPeriod.MONTHLY, NextEvent.CREATE, NextEvent.INVOICE);
         // Check bundle after BP got created otherwise we get an error from auditApi.
         subscriptionChecker.checkSubscriptionCreated(bpSubscription.getId(), internalCallContext);

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/caching/EhCacheCatalogCache.java b/catalog/src/main/java/org/killbill/billing/catalog/caching/EhCacheCatalogCache.java
index b43e99d..8f31169 100644
--- a/catalog/src/main/java/org/killbill/billing/catalog/caching/EhCacheCatalogCache.java
+++ b/catalog/src/main/java/org/killbill/billing/catalog/caching/EhCacheCatalogCache.java
@@ -70,8 +70,14 @@ public class EhCacheCatalogCache implements CatalogCache {
         // The cache loader might choke on some bad xml -- unlikely since we check its validity prior storing it,
         // but to be on the safe side;;
         try {
-            final VersionedCatalog tenantCatalog = (VersionedCatalog) cacheController.get(tenantContext.getTenantRecordId(), cacheLoaderArgument);
-            return (tenantCatalog != null) ? tenantCatalog : defaultCatalog;
+            VersionedCatalog tenantCatalog = (VersionedCatalog) cacheController.get(tenantContext.getTenantRecordId(), cacheLoaderArgument);
+            // It means we are using a default catalog in a multi-tenant deployment, that does not really match a real use case, but we want to support it
+            // for test purpose.
+            if (tenantCatalog == null) {
+                tenantCatalog = new VersionedCatalog(defaultCatalog, tenantContext);
+                cacheController.add(tenantContext.getTenantRecordId(), tenantCatalog);
+            }
+            return tenantCatalog;
         } catch (final IllegalStateException e) {
             throw new CatalogApiException(ErrorCode.CAT_INVALID_FOR_TENANT, tenantContext.getTenantRecordId());
         }

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/caching/EhCacheOverriddenPlanCache.java b/catalog/src/main/java/org/killbill/billing/catalog/caching/EhCacheOverriddenPlanCache.java
new file mode 100644
index 0000000..5e94a51
--- /dev/null
+++ b/catalog/src/main/java/org/killbill/billing/catalog/caching/EhCacheOverriddenPlanCache.java
@@ -0,0 +1,113 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project 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 org.killbill.billing.catalog.caching;
+
+import java.util.List;
+import java.util.regex.Matcher;
+
+import javax.inject.Inject;
+
+import org.killbill.billing.ErrorCode;
+import org.killbill.billing.ObjectType;
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.catalog.DefaultPlan;
+import org.killbill.billing.catalog.DefaultPlanPhasePriceOverride;
+import org.killbill.billing.catalog.api.CatalogApiException;
+import org.killbill.billing.catalog.api.Currency;
+import org.killbill.billing.catalog.api.Plan;
+import org.killbill.billing.catalog.api.PlanPhase;
+import org.killbill.billing.catalog.api.PlanPhasePriceOverride;
+import org.killbill.billing.catalog.api.StaticCatalog;
+import org.killbill.billing.catalog.dao.CatalogOverrideDao;
+import org.killbill.billing.catalog.dao.CatalogOverridePhaseDefinitionModelDao;
+import org.killbill.billing.catalog.override.DefaultPriceOverride;
+import org.killbill.billing.util.cache.Cachable.CacheType;
+import org.killbill.billing.util.cache.CacheController;
+import org.killbill.billing.util.cache.CacheControllerDispatcher;
+import org.killbill.billing.util.cache.CacheLoaderArgument;
+import org.killbill.billing.util.cache.OverriddenPlanCacheLoader.LoaderCallback;
+
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+
+public class EhCacheOverriddenPlanCache implements OverriddenPlanCache {
+
+    private final CacheController cacheController;
+    private final LoaderCallback loaderCallback;
+    private final CatalogOverrideDao overrideDao;
+
+    @Inject
+    public EhCacheOverriddenPlanCache(final CatalogOverrideDao overrideDao, final CacheControllerDispatcher cacheControllerDispatcher) {
+        this.overrideDao = overrideDao;
+        this.cacheController = cacheControllerDispatcher.getCacheController(CacheType.OVERRIDDEN_PLAN);
+        this.loaderCallback = new LoaderCallback() {
+            @Override
+            public Object loadPlan(final String planName, final StaticCatalog catalog, final InternalTenantContext context) throws CatalogApiException {
+                return loadOverriddenPlan(planName, catalog, context);
+            }
+        };
+    }
+
+    @Override
+    public DefaultPlan getOverriddenPlan(final String planName, final StaticCatalog catalog, final InternalTenantContext context) {
+
+        final ObjectType irrelevant = null;
+        final Object[] args = new Object[2];
+        args[0] = loaderCallback;
+        args[1] = catalog;
+
+        final CacheLoaderArgument argument = new CacheLoaderArgument(irrelevant, args, context);
+        return (DefaultPlan) cacheController.get(planName, argument);
+    }
+
+    private DefaultPlan loadOverriddenPlan(final String planName, final StaticCatalog catalog, final InternalTenantContext context) throws CatalogApiException {
+
+        final Matcher m = DefaultPriceOverride.CUSTOM_PLAN_NAME_PATTERN.matcher(planName);
+        if (!m.matches()) {
+            throw new CatalogApiException(ErrorCode.CAT_NO_SUCH_PLAN, planName);
+        }
+        final String parentPlanName = m.group(1);
+        final Long planDefRecordId = Long.parseLong(m.group(2));
+
+        final List<CatalogOverridePhaseDefinitionModelDao> phaseDefs = overrideDao.getOverriddenPlanPhases(planDefRecordId, context);
+        final DefaultPlan defaultPlan = (DefaultPlan) catalog.findCurrentPlan(parentPlanName);
+
+        final PlanPhasePriceOverride[] overrides = createOverrides(defaultPlan, phaseDefs);
+        return new DefaultPlan(planName, defaultPlan, overrides);
+    }
+
+    private PlanPhasePriceOverride[] createOverrides(final Plan defaultPlan, final List<CatalogOverridePhaseDefinitionModelDao> phaseDefs) {
+
+        final PlanPhasePriceOverride[] result = new PlanPhasePriceOverride[defaultPlan.getAllPhases().length];
+
+        for (int i = 0; i < defaultPlan.getAllPhases().length; i++) {
+
+            final PlanPhase curPhase = defaultPlan.getAllPhases()[i];
+            final CatalogOverridePhaseDefinitionModelDao overriddenPhase = Iterables.tryFind(phaseDefs, new Predicate<CatalogOverridePhaseDefinitionModelDao>() {
+                @Override
+                public boolean apply(final CatalogOverridePhaseDefinitionModelDao input) {
+                    return input.getParentPhaseName().equals(curPhase.getName());
+                }
+            }).orNull();
+            result[i] = (overriddenPhase != null) ?
+                        new DefaultPlanPhasePriceOverride(curPhase.getName(), Currency.valueOf(overriddenPhase.getCurrency()), overriddenPhase.getFixedPrice(), overriddenPhase.getRecurringPrice()) :
+                        null;
+        }
+        return result;
+    }
+}

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/caching/OverriddenPlanCache.java b/catalog/src/main/java/org/killbill/billing/catalog/caching/OverriddenPlanCache.java
new file mode 100644
index 0000000..f0f8178
--- /dev/null
+++ b/catalog/src/main/java/org/killbill/billing/catalog/caching/OverriddenPlanCache.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project 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 org.killbill.billing.catalog.caching;
+
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.catalog.DefaultPlan;
+import org.killbill.billing.catalog.api.CatalogApiException;
+import org.killbill.billing.catalog.api.StaticCatalog;
+
+public interface OverriddenPlanCache {
+
+    DefaultPlan getOverriddenPlan(final String planName, final StaticCatalog catalog, final InternalTenantContext context) throws CatalogApiException;
+}

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/glue/CatalogModule.java b/catalog/src/main/java/org/killbill/billing/catalog/glue/CatalogModule.java
index 32685e9..4e93f86 100644
--- a/catalog/src/main/java/org/killbill/billing/catalog/glue/CatalogModule.java
+++ b/catalog/src/main/java/org/killbill/billing/catalog/glue/CatalogModule.java
@@ -25,6 +25,8 @@ import org.killbill.billing.catalog.api.user.DefaultCatalogUserApi;
 import org.killbill.billing.catalog.caching.CatalogCache;
 import org.killbill.billing.catalog.caching.CatalogCacheInvalidationCallback;
 import org.killbill.billing.catalog.caching.EhCacheCatalogCache;
+import org.killbill.billing.catalog.caching.EhCacheOverriddenPlanCache;
+import org.killbill.billing.catalog.caching.OverriddenPlanCache;
 import org.killbill.billing.catalog.dao.CatalogOverrideDao;
 import org.killbill.billing.catalog.dao.DefaultCatalogOverrideDao;
 import org.killbill.billing.catalog.io.CatalogLoader;
@@ -69,6 +71,8 @@ public class CatalogModule extends KillBillModule {
     public void installCatalogConfigCache() {
         bind(CatalogCache.class).to(EhCacheCatalogCache.class).asEagerSingleton();
         bind(CacheInvalidationCallback.class).annotatedWith(Names.named(CATALOG_INVALIDATION_CALLBACK)).to(CatalogCacheInvalidationCallback.class).asEagerSingleton();
+
+        bind(OverriddenPlanCache.class).to(EhCacheOverriddenPlanCache.class).asEagerSingleton();
     }
 
 

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/io/VersionedCatalogLoader.java b/catalog/src/main/java/org/killbill/billing/catalog/io/VersionedCatalogLoader.java
index 632d393..5d0e820 100644
--- a/catalog/src/main/java/org/killbill/billing/catalog/io/VersionedCatalogLoader.java
+++ b/catalog/src/main/java/org/killbill/billing/catalog/io/VersionedCatalogLoader.java
@@ -82,7 +82,7 @@ public class VersionedCatalogLoader implements CatalogLoader {
                 xmlURIs = findXmlReferences(directoryContents, new URL(uriString));
             }
 
-            final VersionedCatalog result = new VersionedCatalog(clock, InternalCallContextFactory.INTERNAL_TENANT_RECORD_ID);
+            final VersionedCatalog result = new VersionedCatalog(clock);
             for (final URI u : xmlURIs) {
                 final StandaloneCatalog catalog = XMLLoader.getObjectFromUri(u, StandaloneCatalog.class);
                 result.add(new StandaloneCatalogWithPriceOverride(catalog, priceOverride, InternalCallContextFactory.INTERNAL_TENANT_RECORD_ID, internalCallContextFactory));
@@ -94,7 +94,7 @@ public class VersionedCatalogLoader implements CatalogLoader {
     }
 
     public VersionedCatalog load(final List<String> catalogXMLs, final Long tenantRecordId) throws CatalogApiException {
-        final VersionedCatalog result = new VersionedCatalog(clock, tenantRecordId);
+        final VersionedCatalog result = new VersionedCatalog(clock);
         final URI uri;
         try {
             uri = new URI("/tenantCatalog");

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/override/DefaultPriceOverride.java b/catalog/src/main/java/org/killbill/billing/catalog/override/DefaultPriceOverride.java
index 897af0e..c8165b5 100644
--- a/catalog/src/main/java/org/killbill/billing/catalog/override/DefaultPriceOverride.java
+++ b/catalog/src/main/java/org/killbill/billing/catalog/override/DefaultPriceOverride.java
@@ -18,7 +18,6 @@
 package org.killbill.billing.catalog.override;
 
 import java.util.List;
-import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 import org.joda.time.DateTime;
@@ -29,14 +28,13 @@ import org.killbill.billing.catalog.DefaultPlan;
 import org.killbill.billing.catalog.DefaultPlanPhase;
 import org.killbill.billing.catalog.DefaultPlanPhasePriceOverride;
 import org.killbill.billing.catalog.api.CatalogApiException;
-import org.killbill.billing.catalog.api.Currency;
 import org.killbill.billing.catalog.api.Plan;
 import org.killbill.billing.catalog.api.PlanPhase;
 import org.killbill.billing.catalog.api.PlanPhasePriceOverride;
 import org.killbill.billing.catalog.api.PlanPhaseSpecifier;
 import org.killbill.billing.catalog.api.StaticCatalog;
+import org.killbill.billing.catalog.caching.OverriddenPlanCache;
 import org.killbill.billing.catalog.dao.CatalogOverrideDao;
-import org.killbill.billing.catalog.dao.CatalogOverridePhaseDefinitionModelDao;
 import org.killbill.billing.catalog.dao.CatalogOverridePlanDefinitionModelDao;
 
 import com.google.common.base.Predicate;
@@ -48,10 +46,12 @@ public class DefaultPriceOverride implements PriceOverride {
     public static final Pattern CUSTOM_PLAN_NAME_PATTERN = Pattern.compile("(.*)-(\\d+)$");
 
     private final CatalogOverrideDao overrideDao;
+    private final OverriddenPlanCache overriddenPlanCache;
 
     @Inject
-    public DefaultPriceOverride(final CatalogOverrideDao overrideDao) {
+    public DefaultPriceOverride(final CatalogOverrideDao overrideDao, final OverriddenPlanCache overriddenPlanCache) {
         this.overrideDao = overrideDao;
+        this.overriddenPlanCache = overriddenPlanCache;
     }
 
     @Override
@@ -66,7 +66,7 @@ public class DefaultPriceOverride implements PriceOverride {
                     if (input.getPhaseName() != null) {
                         return input.getPhaseName().equals(curPhase.getName());
                     }
-                    // If the phaseName was not passed, we infer by matching the phaseType. This obvously would not work in a case where
+                    // If the phaseName was not passed, we infer by matching the phaseType. This obviously would not work in a case where
                     // a plan is defined with multiple phases of the same type.
                     final PlanPhaseSpecifier curPlanPhaseSpecifier = input.getPlanPhaseSpecifier();
                     if (curPlanPhaseSpecifier.getPhaseType().equals(curPhase.getPhaseType())) {
@@ -105,39 +105,6 @@ public class DefaultPriceOverride implements PriceOverride {
 
     @Override
     public DefaultPlan getOverriddenPlan(final String planName, final StaticCatalog catalog, final InternalTenantContext context) throws CatalogApiException {
-
-        final Matcher m = CUSTOM_PLAN_NAME_PATTERN.matcher(planName);
-        if (!m.matches()) {
-            throw new CatalogApiException(ErrorCode.CAT_NO_SUCH_PLAN, planName);
-        }
-        final String parentPlanName = m.group(1);
-        final Long planDefRecordId = Long.parseLong(m.group(2));
-
-        final List<CatalogOverridePhaseDefinitionModelDao> phaseDefs = overrideDao.getOverriddenPlanPhases(planDefRecordId, context);
-        final DefaultPlan defaultPlan = (DefaultPlan) catalog.findCurrentPlan(parentPlanName);
-
-        final PlanPhasePriceOverride[] overrides = createOverrides(defaultPlan, phaseDefs);
-        return new DefaultPlan(planName, defaultPlan, overrides);
+        return overriddenPlanCache.getOverriddenPlan(planName, catalog, context);
     }
-
-    private PlanPhasePriceOverride[] createOverrides(final Plan defaultPlan, final List<CatalogOverridePhaseDefinitionModelDao> phaseDefs) {
-
-        final PlanPhasePriceOverride[] result = new PlanPhasePriceOverride[defaultPlan.getAllPhases().length];
-
-        for (int i = 0; i < defaultPlan.getAllPhases().length; i++) {
-
-            final PlanPhase curPhase = defaultPlan.getAllPhases()[i];
-            final CatalogOverridePhaseDefinitionModelDao overriddenPhase = Iterables.tryFind(phaseDefs, new Predicate<CatalogOverridePhaseDefinitionModelDao>() {
-                @Override
-                public boolean apply(final CatalogOverridePhaseDefinitionModelDao input) {
-                    return input.getParentPhaseName().equals(curPhase.getName());
-                }
-            }).orNull();
-            result[i] = (overriddenPhase != null) ?
-                        new DefaultPlanPhasePriceOverride(curPhase.getName(), Currency.valueOf(overriddenPhase.getCurrency()), overriddenPhase.getFixedPrice(), overriddenPhase.getRecurringPrice()) :
-                        null;
-        }
-        return result;
-    }
-
 }

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/StandaloneCatalogWithPriceOverride.java b/catalog/src/main/java/org/killbill/billing/catalog/StandaloneCatalogWithPriceOverride.java
index 11b1bfe..ac2326a 100644
--- a/catalog/src/main/java/org/killbill/billing/catalog/StandaloneCatalogWithPriceOverride.java
+++ b/catalog/src/main/java/org/killbill/billing/catalog/StandaloneCatalogWithPriceOverride.java
@@ -52,13 +52,18 @@ import org.killbill.billing.util.callcontext.InternalCallContextFactory;
 import org.killbill.xmlloader.ValidatingConfig;
 import org.killbill.xmlloader.ValidationErrors;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
 public class StandaloneCatalogWithPriceOverride extends ValidatingConfig<StandaloneCatalogWithPriceOverride> implements StaticCatalog {
 
     private final StandaloneCatalog standaloneCatalog;
-    private final PriceOverride priceOverride;
     private final Long tenantRecordId;
 
+    /* Since we offer endpoints that attempt to serialize catalog objects, we need to explicitly tell Jackson to ignore those fields */
+    @JsonIgnore
     private final InternalCallContextFactory internalCallContextFactory;
+    @JsonIgnore
+    private final PriceOverride priceOverride;
 
     public StandaloneCatalogWithPriceOverride(final StandaloneCatalog staticCatalog, final PriceOverride priceOverride, final Long tenantRecordId, final InternalCallContextFactory internalCallContextFactory) {
         this.tenantRecordId = tenantRecordId;
@@ -67,6 +72,29 @@ public class StandaloneCatalogWithPriceOverride extends ValidatingConfig<Standal
         this.internalCallContextFactory = internalCallContextFactory;
     }
 
+    public StandaloneCatalogWithPriceOverride(final StandaloneCatalogWithPriceOverride cur, final InternalTenantContext tenantContext) {
+        this.tenantRecordId = tenantContext.getTenantRecordId();
+        this.priceOverride = cur.getPriceOverride();
+        this.standaloneCatalog = cur.getStandaloneCatalog();
+        this.internalCallContextFactory = cur.getInternalCallContextFactory();
+    }
+
+    public StandaloneCatalog getStandaloneCatalog() {
+        return standaloneCatalog;
+    }
+
+    public PriceOverride getPriceOverride() {
+        return priceOverride;
+    }
+
+    public Long getTenantRecordId() {
+        return tenantRecordId;
+    }
+
+    public InternalCallContextFactory getInternalCallContextFactory() {
+        return internalCallContextFactory;
+    }
+
     @Override
     public String getCatalogName() {
         return standaloneCatalog.getCatalogName();

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/VersionedCatalog.java b/catalog/src/main/java/org/killbill/billing/catalog/VersionedCatalog.java
index 77de272..38e9181 100644
--- a/catalog/src/main/java/org/killbill/billing/catalog/VersionedCatalog.java
+++ b/catalog/src/main/java/org/killbill/billing/catalog/VersionedCatalog.java
@@ -34,6 +34,7 @@ import javax.xml.bind.annotation.XmlRootElement;
 
 import org.joda.time.DateTime;
 import org.killbill.billing.ErrorCode;
+import org.killbill.billing.callcontext.InternalTenantContext;
 import org.killbill.billing.catalog.api.BillingActionPolicy;
 import org.killbill.billing.catalog.api.BillingAlignment;
 import org.killbill.billing.catalog.api.BillingMode;
@@ -66,7 +67,6 @@ public class VersionedCatalog extends ValidatingConfig<StandaloneCatalogWithPric
     private final Clock clock;
     private String catalogName;
     private BillingMode recurringBillingMode;
-    private final Long tenantRecordId;
 
     @XmlElement(name = "catalogVersion", required = true)
     private final List<StandaloneCatalogWithPriceOverride> versions = new ArrayList<StandaloneCatalogWithPriceOverride>();
@@ -74,14 +74,21 @@ public class VersionedCatalog extends ValidatingConfig<StandaloneCatalogWithPric
     // Required for JAXB deserialization
     public VersionedCatalog() {
         this.clock = null;
-        this.tenantRecordId = null;
     }
 
-    public VersionedCatalog(final Clock clock, final Long tenantRecordId) {
+    public VersionedCatalog(final Clock clock) {
         this.clock = clock;
-        this.tenantRecordId = tenantRecordId;
     }
 
+    public VersionedCatalog(final VersionedCatalog inputCatalog, final InternalTenantContext tenantContext) {
+        this.clock = inputCatalog.getClock();
+        this.catalogName = inputCatalog.getCatalogName();
+        this.recurringBillingMode = inputCatalog.getRecurringBillingMode();
+        for (final StandaloneCatalogWithPriceOverride cur : inputCatalog.getVersions()) {
+            final StandaloneCatalogWithPriceOverride catalogWithTenantInfo = new StandaloneCatalogWithPriceOverride(cur, tenantContext);
+            versions.add(catalogWithTenantInfo);
+        }
+    }
 
     //
     // Private methods
@@ -177,6 +184,14 @@ public class VersionedCatalog extends ValidatingConfig<StandaloneCatalogWithPric
         throw new CatalogApiException(ErrorCode.CAT_NO_CATALOG_FOR_GIVEN_DATE, requestedDate.toDate().toString());
     }
 
+    public Clock getClock() {
+        return clock;
+    }
+
+    public List<StandaloneCatalogWithPriceOverride> getVersions() {
+        return versions;
+    }
+
     //
     // Public methods not exposed in interface
     //

diff --git a/catalog/src/main/resources/org/killbill/billing/catalog/dao/CatalogOverridePlanPhaseSqlDao.sql.stg b/catalog/src/main/resources/org/killbill/billing/catalog/dao/CatalogOverridePlanPhaseSqlDao.sql.stg
index 356eea2..be35a86 100644
--- a/catalog/src/main/resources/org/killbill/billing/catalog/dao/CatalogOverridePlanPhaseSqlDao.sql.stg
+++ b/catalog/src/main/resources/org/killbill/billing/catalog/dao/CatalogOverridePlanPhaseSqlDao.sql.stg
@@ -64,7 +64,7 @@ from (select
       where
       concat_ws(',', phase_number, phase_def_record_id) in (<keys: {key | :key_<i0>}; separator="," >)
       and tenant_record_id = :tenantRecordId
-      group by 1) tmp
+      group by target_plan_def_record_id) tmp
 where
 1=1
 and tmp.count = :targetCount

diff --git a/catalog/src/test/java/org/killbill/billing/catalog/caching/TestEhCacheCatalogCache.java b/catalog/src/test/java/org/killbill/billing/catalog/caching/TestEhCacheCatalogCache.java
index 611378c..998f2ae 100644
--- a/catalog/src/test/java/org/killbill/billing/catalog/caching/TestEhCacheCatalogCache.java
+++ b/catalog/src/test/java/org/killbill/billing/catalog/caching/TestEhCacheCatalogCache.java
@@ -83,10 +83,12 @@ public class TestEhCacheCatalogCache extends CatalogTestSuiteNoDB {
         Assert.assertEquals(products.length, 3);
 
         // Verify the lookup with other contexts
-        Assert.assertEquals(catalogCache.getCatalog(multiTenantContext), result);
-        Assert.assertEquals(catalogCache.getCatalog(otherMultiTenantContext), result);
-        Assert.assertEquals(catalogCache.getCatalog(Mockito.mock(InternalTenantContext.class)), result);
-        Assert.assertEquals(catalogCache.getCatalog(Mockito.mock(InternalCallContext.class)), result);
+        final VersionedCatalog resultForMultiTenantContext = new VersionedCatalog(result, multiTenantContext);
+        Assert.assertEquals(catalogCache.getCatalog(multiTenantContext).getCatalogName(), resultForMultiTenantContext.getCatalogName());
+        Assert.assertEquals(catalogCache.getCatalog(multiTenantContext).getVersions().size(), resultForMultiTenantContext.getVersions().size());
+        for (int i = 0; i < catalogCache.getCatalog(multiTenantContext).getVersions().size(); i++) {
+            Assert.assertEquals(catalogCache.getCatalog(multiTenantContext).getVersions().get(i).getTenantRecordId(), resultForMultiTenantContext.getVersions().get(i).getTenantRecordId());
+        }
     }
 
     //

diff --git a/jaxrs/src/main/java/org/killbill/billing/jaxrs/json/SubscriptionJson.java b/jaxrs/src/main/java/org/killbill/billing/jaxrs/json/SubscriptionJson.java
index e0ddda5..fab4d96 100644
--- a/jaxrs/src/main/java/org/killbill/billing/jaxrs/json/SubscriptionJson.java
+++ b/jaxrs/src/main/java/org/killbill/billing/jaxrs/json/SubscriptionJson.java
@@ -333,8 +333,8 @@ public class SubscriptionJson extends JsonBase {
         for (final SubscriptionEvent subscriptionEvent : subscription.getSubscriptionEvents()) {
             this.events.add(new EventSubscriptionJson(subscriptionEvent, accountAuditLogs));
         }
+        // It may be nice to recreate the override that were applied on the plans associated with that subscription
         this.priceOverrides = new LinkedList<PhasePriceOverrideJson>();
-        // STEPH_PO
     }
 
     public String getAccountId() {

diff --git a/profiles/killbill/src/test/java/org/killbill/billing/jaxrs/KillbillClient.java b/profiles/killbill/src/test/java/org/killbill/billing/jaxrs/KillbillClient.java
index bfcd862..3686716 100644
--- a/profiles/killbill/src/test/java/org/killbill/billing/jaxrs/KillbillClient.java
+++ b/profiles/killbill/src/test/java/org/killbill/billing/jaxrs/KillbillClient.java
@@ -40,7 +40,7 @@ import static org.testng.Assert.assertNotNull;
 
 public abstract class KillbillClient extends GuicyKillbillTestSuiteWithEmbeddedDB {
 
-    private final int DEFAULT_WAIT_COMPLETION_TIMEOUT_SEC = 5;
+    protected final int DEFAULT_WAIT_COMPLETION_TIMEOUT_SEC = 5;
 
     protected static final String PLUGIN_NAME = "noop";
 

diff --git a/profiles/killbill/src/test/java/org/killbill/billing/jaxrs/TestEntitlement.java b/profiles/killbill/src/test/java/org/killbill/billing/jaxrs/TestEntitlement.java
index 39c2c7c..875e8aa 100644
--- a/profiles/killbill/src/test/java/org/killbill/billing/jaxrs/TestEntitlement.java
+++ b/profiles/killbill/src/test/java/org/killbill/billing/jaxrs/TestEntitlement.java
@@ -18,6 +18,9 @@
 
 package org.killbill.billing.jaxrs;
 
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.UUID;
 
 import org.joda.time.DateTime;
@@ -25,12 +28,17 @@ import org.joda.time.Interval;
 import org.joda.time.LocalDate;
 import org.killbill.billing.catalog.api.BillingActionPolicy;
 import org.killbill.billing.catalog.api.BillingPeriod;
+import org.killbill.billing.catalog.api.PhaseType;
 import org.killbill.billing.catalog.api.PriceListSet;
 import org.killbill.billing.catalog.api.ProductCategory;
+import org.killbill.billing.catalog.override.PriceOverride;
 import org.killbill.billing.client.KillBillClientException;
 import org.killbill.billing.client.model.Account;
+import org.killbill.billing.client.model.Invoice;
+import org.killbill.billing.client.model.PhasePriceOverride;
 import org.killbill.billing.client.model.Subscription;
 import org.killbill.billing.entitlement.api.Entitlement.EntitlementActionPolicy;
+import org.killbill.billing.util.api.AuditLevel;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
@@ -189,4 +197,35 @@ public class TestEntitlement extends TestJaxrsBase {
         Assert.assertNotNull(objFromJson);
         assertEquals(objFromJson.getBillingPeriod(), BillingPeriod.MONTHLY);
     }
+
+
+    @Test(groups = "slow", description = "Can override a price when creating a subscription")
+    public void testOverridePrice() throws Exception {
+        final DateTime initialDate = new DateTime(2012, 4, 25, 0, 3, 42, 0);
+        clock.setDeltaFromReality(initialDate.getMillis() - clock.getUTCNow().getMillis());
+
+        final Account accountJson = createAccountWithDefaultPaymentMethod();
+
+        final String productName = "Shotgun";
+        final BillingPeriod term = BillingPeriod.ANNUAL;
+
+
+        final Subscription input = new Subscription();
+        input.setAccountId(accountJson.getAccountId());
+        input.setExternalKey("identical");
+        input.setProductName(productName);
+        input.setProductCategory(ProductCategory.BASE);
+        input.setBillingPeriod(BillingPeriod.MONTHLY);
+        input.setPriceList(PriceListSet.DEFAULT_PRICELIST_NAME);
+        final List<PhasePriceOverride> overrides = new ArrayList<PhasePriceOverride>();
+        overrides.add(new PhasePriceOverride(null, PhaseType.TRIAL.toString(), BigDecimal.TEN, null));
+        input.setPriceOverrides(overrides);
+
+        final Subscription subscription =  killBillClient.createSubscription(input,  DEFAULT_WAIT_COMPLETION_TIMEOUT_SEC, createdBy, reason, comment);
+
+        final List<Invoice> invoices = killBillClient.getInvoicesForAccount(accountJson.getAccountId(), true, false, AuditLevel.FULL);
+        assertEquals(invoices.size(), 1);
+        assertEquals(invoices.get(0).getAmount().compareTo(BigDecimal.TEN), 0);
+    }
+
 }

diff --git a/util/src/main/java/org/killbill/billing/util/cache/Cachable.java b/util/src/main/java/org/killbill/billing/util/cache/Cachable.java
index 542b14e..767cc94 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/Cachable.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/Cachable.java
@@ -34,10 +34,12 @@ public @interface Cachable {
     public final String TENANT_CATALOG_CACHE_NAME = "tenant-catalog";
     public final String TENANT_OVERDUE_CONFIG_CACHE_NAME = "tenant-overdue-config";
     public final String TENANT_KV_CACHE_NAME = "tenant-kv";
+    public final String OVERRIDDEN_PLAN_CACHE_NAME = "overridden-plan";
 
     public CacheType value();
 
     public enum CacheType {
+
         /* Mapping from object 'id (UUID)' -> object 'recordId (Long' */
         RECORD_ID(RECORD_ID_CACHE_NAME, false),
 
@@ -63,7 +65,10 @@ public @interface Cachable {
         TENANT_OVERDUE_CONFIG(TENANT_OVERDUE_CONFIG_CACHE_NAME, false),
 
         /* Tenant overdue config cache */
-        TENANT_KV(TENANT_KV_CACHE_NAME, false);
+        TENANT_KV(TENANT_KV_CACHE_NAME, false),
+
+        /* Overwritten plans  */
+        OVERRIDDEN_PLAN(OVERRIDDEN_PLAN_CACHE_NAME, false);
 
         private final String cacheName;
         private final boolean isKeyPrefixedWithTableName;

diff --git a/util/src/main/java/org/killbill/billing/util/cache/EhCacheCacheManagerProvider.java b/util/src/main/java/org/killbill/billing/util/cache/EhCacheCacheManagerProvider.java
index b632255..42d8966 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/EhCacheCacheManagerProvider.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/EhCacheCacheManagerProvider.java
@@ -59,7 +59,8 @@ public class EhCacheCacheManagerProvider implements Provider<CacheManager> {
                                        final AuditLogViaHistoryCacheLoader auditLogViaHistoryCacheLoader,
                                        final TenantCatalogCacheLoader tenantCatalogCacheLoader,
                                        final TenantOverdueConfigCacheLoader tenantOverdueConfigCacheLoader,
-                                       final TenantKVCacheLoader tenantKVCacheLoader) {
+                                       final TenantKVCacheLoader tenantKVCacheLoader,
+                                       final OverriddenPlanCacheLoader overriddenPlanCacheLoader) {
         this.metricRegistry = metricRegistry;
         this.cacheConfig = cacheConfig;
         cacheLoaders.add(recordIdCacheLoader);
@@ -71,6 +72,7 @@ public class EhCacheCacheManagerProvider implements Provider<CacheManager> {
         cacheLoaders.add(tenantCatalogCacheLoader);
         cacheLoaders.add(tenantOverdueConfigCacheLoader);
         cacheLoaders.add(tenantKVCacheLoader);
+        cacheLoaders.add(overriddenPlanCacheLoader);
     }
 
     @Override

diff --git a/util/src/main/java/org/killbill/billing/util/cache/OverriddenPlanCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/OverriddenPlanCacheLoader.java
new file mode 100644
index 0000000..ab2e833
--- /dev/null
+++ b/util/src/main/java/org/killbill/billing/util/cache/OverriddenPlanCacheLoader.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project 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 org.killbill.billing.util.cache;
+
+import java.util.List;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.catalog.api.CatalogApiException;
+import org.killbill.billing.catalog.api.StaticCatalog;
+import org.killbill.billing.tenant.api.TenantInternalApi;
+import org.killbill.billing.util.cache.Cachable.CacheType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Singleton
+public class OverriddenPlanCacheLoader extends BaseCacheLoader {
+
+    private final Logger log = LoggerFactory.getLogger(OverriddenPlanCacheLoader.class);
+
+
+    @Inject
+    public OverriddenPlanCacheLoader(final TenantInternalApi tenantApi) {
+        super();
+    }
+
+    @Override
+    public CacheType getCacheType() {
+        return CacheType.OVERRIDDEN_PLAN;
+    }
+
+    @Override
+    public Object load(final Object key, final Object argument) {
+        checkCacheLoaderStatus();
+
+        if (!(key instanceof String)) {
+            throw new IllegalArgumentException("Unexpected key type of " + key.getClass().getName());
+        }
+        if (!(argument instanceof CacheLoaderArgument)) {
+            throw new IllegalArgumentException("Unexpected argument type of " + argument.getClass().getName());
+        }
+
+
+        final CacheLoaderArgument cacheLoaderArgument = (CacheLoaderArgument) argument;
+        if (cacheLoaderArgument.getArgs() == null || cacheLoaderArgument.getArgs().length != 2) {
+            throw new IllegalArgumentException("Invalid arguments for overridden plans");
+        }
+        if (!(cacheLoaderArgument.getArgs()[0] instanceof LoaderCallback)) {
+            throw new IllegalArgumentException("Invalid arguments for overridden plans: missing loaderCallback from argument");
+        }
+
+        if (!(cacheLoaderArgument.getArgs()[1] instanceof StaticCatalog)) {
+            throw new IllegalArgumentException("Invalid arguments for overridden plans: missing catalog from argument");
+        }
+
+
+        final String planName = (String) key;
+        final LoaderCallback callback = (LoaderCallback) cacheLoaderArgument.getArgs()[0];
+        final StaticCatalog catalog = (StaticCatalog) cacheLoaderArgument.getArgs()[1];
+        final InternalTenantContext internalTenantContext = ((CacheLoaderArgument) argument).getInternalTenantContext();
+        try {
+            log.info("Loading overridden plan {} for tenant {}", planName, internalTenantContext.getTenantRecordId());
+
+            return callback.loadPlan(planName, catalog, internalTenantContext);
+        } catch (final CatalogApiException e) {
+            throw new IllegalStateException(String.format("Failed to load overridden plan for tenant %s : %s",
+                                                          planName, internalTenantContext.getTenantRecordId()), e);
+        }
+    }
+
+    public interface LoaderCallback {
+        public Object loadPlan(final String planName, final StaticCatalog catalog, final InternalTenantContext context) throws CatalogApiException;
+    }
+}

diff --git a/util/src/main/resources/ehcache.xml b/util/src/main/resources/ehcache.xml
index bf0d8cb..7769245 100644
--- a/util/src/main/resources/ehcache.xml
+++ b/util/src/main/resources/ehcache.xml
@@ -155,5 +155,19 @@
                 properties=""/>
     </cache>
 
+    <cache name="overridden-plan"
+           maxElementsInMemory="1000"
+           maxElementsOnDisk="0"
+           overflowToDisk="false"
+           diskPersistent="false"
+           memoryStoreEvictionPolicy="LFU"
+           statistics="true"
+            >
+        <cacheEventListenerFactory
+                class="org.killbill.billing.util.cache.ExpirationListenerFactory"
+                properties=""/>
+    </cache>
+
+
 </ehcache>