killbill-uncached

diff --git a/account/src/test/java/org/killbill/billing/account/glue/TestAccountModule.java b/account/src/test/java/org/killbill/billing/account/glue/TestAccountModule.java
index 8243902..044e4f9 100644
--- a/account/src/test/java/org/killbill/billing/account/glue/TestAccountModule.java
+++ b/account/src/test/java/org/killbill/billing/account/glue/TestAccountModule.java
@@ -19,6 +19,7 @@
 package org.killbill.billing.account.glue;
 
 import org.killbill.billing.mock.glue.MockSubscriptionModule;
+import org.killbill.billing.mock.glue.MockTenantModule;
 import org.killbill.billing.platform.api.KillbillConfigSource;
 import org.killbill.billing.util.glue.AuditModule;
 import org.killbill.billing.util.glue.CacheModule;
@@ -40,6 +41,7 @@ public class TestAccountModule extends DefaultAccountModule {
         install(new CacheModule(configSource));
         install(new CallContextModule(configSource));
         install(new CustomFieldModule(configSource));
+        install(new MockTenantModule(configSource));
         // Needed for Audit
         install(new MockSubscriptionModule(configSource));
         install(new TagStoreModule(configSource));

diff --git a/api/src/main/java/org/killbill/billing/tenant/api/TenantInternalApi.java b/api/src/main/java/org/killbill/billing/tenant/api/TenantInternalApi.java
index 08ca044..2b7aee0 100644
--- a/api/src/main/java/org/killbill/billing/tenant/api/TenantInternalApi.java
+++ b/api/src/main/java/org/killbill/billing/tenant/api/TenantInternalApi.java
@@ -22,5 +22,5 @@ import java.util.List;
 import org.killbill.billing.callcontext.InternalTenantContext;
 
 public interface TenantInternalApi {
-    List<String> getTenantCatalogs(InternalTenantContext tenantContext) throws TenantApiException;
+    List<String> getTenantCatalogs(InternalTenantContext tenantContext);
 }

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/caching/CatalogCache.java b/catalog/src/main/java/org/killbill/billing/catalog/caching/CatalogCache.java
new file mode 100644
index 0000000..b7bf7d9
--- /dev/null
+++ b/catalog/src/main/java/org/killbill/billing/catalog/caching/CatalogCache.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2014 Groupon, Inc
+ * Copyright 2014 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.VersionedCatalog;
+import org.killbill.billing.catalog.api.CatalogApiException;
+
+public interface CatalogCache {
+    public void loadDefaultCatalog(final String url) throws CatalogApiException;
+    public VersionedCatalog getCatalog(InternalTenantContext tenantContext) throws CatalogApiException;
+}

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
new file mode 100644
index 0000000..dd0769f
--- /dev/null
+++ b/catalog/src/main/java/org/killbill/billing/catalog/caching/EhCacheCatalogCache.java
@@ -0,0 +1,89 @@
+/*
+ * Copyright 2014 Groupon, Inc
+ * Copyright 2014 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 org.killbill.billing.ErrorCode;
+import org.killbill.billing.ObjectType;
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.catalog.VersionedCatalog;
+import org.killbill.billing.catalog.api.CatalogApiException;
+import org.killbill.billing.catalog.io.VersionedCatalogLoader;
+import org.killbill.billing.util.cache.CacheController;
+import org.killbill.billing.util.cache.CacheLoaderArgument;
+import org.killbill.billing.util.cache.TenantCatalogCacheLoader.LoaderCallback;
+import org.killbill.billing.util.callcontext.InternalCallContextFactory;
+
+public class EhCacheCatalogCache implements CatalogCache {
+
+    private final CacheController cacheController;
+    private final VersionedCatalogLoader loader;
+    private final CacheLoaderArgument cacheLoaderArgument;
+
+    private VersionedCatalog defaultCatalog;
+
+    public EhCacheCatalogCache(final CacheController cacheController, final VersionedCatalogLoader loader) {
+        this.cacheController = cacheController;
+        this.loader = loader;
+        this.cacheLoaderArgument = initializeCacheLoaderArgument();
+    }
+
+    @Override
+    public void loadDefaultCatalog(final String url) throws CatalogApiException {
+        defaultCatalog = loader.load(url);
+    }
+
+    @Override
+    public VersionedCatalog getCatalog(final InternalTenantContext tenantContext) throws CatalogApiException {
+        if (tenantContext.getTenantRecordId() == InternalCallContextFactory.INTERNAL_TENANT_RECORD_ID) {
+            if (defaultCatalog == null) {
+                throw new CatalogApiException(ErrorCode.CAT_INVALID_DEFAULT,
+                                              ": the system property org.killbill.catalog.uri must be specified and point to valid catalog xml");
+            }
+            return defaultCatalog;
+        }
+        // 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, cacheLoaderArgument);
+            return (tenantCatalog != null) ? tenantCatalog : defaultCatalog;
+        } catch (IllegalStateException e) {
+            throw new CatalogApiException(ErrorCode.CAT_INVALID_FOR_TENANT, tenantContext.getTenantRecordId());
+        }
+    }
+
+    //
+    // Build the LoaderCallback that is required to build the catalog from the xml from a module that knows
+    // nothing about catalog.
+    //
+    // This is a contract between the TenantCatalogCacheLoader and the EhCacheCatalogCache
+    private CacheLoaderArgument initializeCacheLoaderArgument() {
+        final LoaderCallback loaderCallback = new LoaderCallback() {
+            @Override
+            public Object loadCatalog(final List<String> catalogXMLs) throws CatalogApiException {
+                return loader.load(catalogXMLs);
+            }
+        };
+        final Object[] args = new Object[1];
+        args[0]= loaderCallback;
+        final ObjectType irrelevant = null;
+        final InternalTenantContext notUsed = null;
+        return new CacheLoaderArgument(irrelevant, args, notUsed);
+    }
+}

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/DefaultCatalogService.java b/catalog/src/main/java/org/killbill/billing/catalog/DefaultCatalogService.java
index 193f4db..6665ac8 100644
--- a/catalog/src/main/java/org/killbill/billing/catalog/DefaultCatalogService.java
+++ b/catalog/src/main/java/org/killbill/billing/catalog/DefaultCatalogService.java
@@ -18,56 +18,53 @@
 
 package org.killbill.billing.catalog;
 
-import java.util.List;
-
-import org.killbill.billing.ErrorCode;
 import org.killbill.billing.callcontext.InternalTenantContext;
 import org.killbill.billing.catalog.api.Catalog;
 import org.killbill.billing.catalog.api.CatalogApiException;
 import org.killbill.billing.catalog.api.CatalogService;
 import org.killbill.billing.catalog.api.StaticCatalog;
+import org.killbill.billing.catalog.caching.CatalogCache;
+import org.killbill.billing.catalog.caching.EhCacheCatalogCache;
 import org.killbill.billing.catalog.io.VersionedCatalogLoader;
 import org.killbill.billing.platform.api.KillbillService;
 import org.killbill.billing.platform.api.LifecycleHandlerType;
 import org.killbill.billing.platform.api.LifecycleHandlerType.LifecycleLevel;
-import org.killbill.billing.tenant.api.TenantApiException;
-import org.killbill.billing.tenant.api.TenantInternalApi;
-import org.killbill.billing.tenant.api.TenantKV.TenantKey;
-import org.killbill.billing.tenant.api.TenantUserApi;
-import org.killbill.billing.util.callcontext.InternalCallContextFactory;
-import org.killbill.billing.util.callcontext.TenantContext;
+import org.killbill.billing.util.cache.Cachable.CacheType;
+import org.killbill.billing.util.cache.CacheControllerDispatcher;
 import org.killbill.billing.util.config.CatalogConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.google.inject.Inject;
-import com.google.inject.Provider;
 
 public class DefaultCatalogService implements KillbillService, CatalogService {
 
+    private static final Logger log = LoggerFactory.getLogger(DefaultCatalogService.class);
     private static final String CATALOG_SERVICE_NAME = "catalog-service";
 
-    private static VersionedCatalog catalog;
-
     private final CatalogConfig config;
     private boolean isInitialized;
 
-    private final VersionedCatalogLoader loader;
-    private final TenantInternalApi tenantApi;
+    private final CatalogCache catalogCache;
 
     @Inject
-    public DefaultCatalogService(final CatalogConfig config, final TenantInternalApi tenantApi, final VersionedCatalogLoader loader) {
+    public DefaultCatalogService(final CatalogConfig config,
+                                 final VersionedCatalogLoader loader,
+                                 final CacheControllerDispatcher cacheControllerDispatcher) {
         this.config = config;
         this.isInitialized = false;
-        this.loader = loader;
-        this.tenantApi = tenantApi;
-
+        this.catalogCache = new EhCacheCatalogCache(cacheControllerDispatcher.getCacheController(CacheType.TENANT_CATALOG), loader);
     }
 
     @LifecycleHandlerType(LifecycleLevel.LOAD_CATALOG)
     public synchronized void loadCatalog() throws ServiceException {
         if (!isInitialized) {
             try {
-                final String url = config.getCatalogURI();
-                catalog = loader.load(url);
+                // In multi-tenant mode, the property is not required
+                if (config.getCatalogURI() != null && !config.getCatalogURI().isEmpty()) {
+                    catalogCache.loadDefaultCatalog(config.getCatalogURI());
+                    log.info("Successfully loaded the default catalog " + config.getCatalogURI());
+                }
                 isInitialized = true;
             } catch (Exception e) {
                 throw new ServiceException(e);
@@ -91,20 +88,6 @@ public class DefaultCatalogService implements KillbillService, CatalogService {
     }
 
     private VersionedCatalog getCatalog(final InternalTenantContext context) throws CatalogApiException {
-        if (context.getTenantRecordId() == InternalCallContextFactory.INTERNAL_TENANT_RECORD_ID) {
-            return catalog;
-        }
-        try {
-            final List<String> catalogXMLs = tenantApi.getTenantCatalogs(context);
-            if (catalogXMLs.isEmpty()) {
-                return catalog;
-            }
-            return loader.load(catalogXMLs);
-        } catch (TenantApiException e) {
-            throw new CatalogApiException(e);
-        } catch (ServiceException e) {
-            throw new CatalogApiException(ErrorCode.CAT_INVALID_FOR_TENANT, "Failed to load catalog for tenant " + context.getTenantRecordId());
-        }
+        return catalogCache.getCatalog(context);
     }
-
 }

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 3326932..033b855 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
@@ -22,7 +22,7 @@ import org.killbill.billing.catalog.DefaultCatalogService;
 import org.killbill.billing.catalog.api.CatalogService;
 import org.killbill.billing.catalog.api.CatalogUserApi;
 import org.killbill.billing.catalog.api.user.DefaultCatalogUserApi;
-import org.killbill.billing.catalog.io.ICatalogLoader;
+import org.killbill.billing.catalog.io.CatalogLoader;
 import org.killbill.billing.catalog.io.VersionedCatalogLoader;
 import org.killbill.billing.platform.api.KillbillConfigSource;
 import org.killbill.billing.util.config.CatalogConfig;
@@ -42,7 +42,7 @@ public class CatalogModule extends KillBillModule {
 
     protected void installCatalog() {
         bind(CatalogService.class).to(DefaultCatalogService.class).asEagerSingleton();
-        bind(ICatalogLoader.class).to(VersionedCatalogLoader.class).asEagerSingleton();
+        bind(CatalogLoader.class).to(VersionedCatalogLoader.class).asEagerSingleton();
     }
 
     protected void installCatalogUserApi() {

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 c45cb40..460b9a6 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
@@ -17,7 +17,6 @@
 package org.killbill.billing.catalog.io;
 
 import java.io.ByteArrayInputStream;
-import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -25,23 +24,19 @@ import java.net.URL;
 import java.util.ArrayList;
 import java.util.List;
 
-import javax.xml.bind.JAXBException;
-import javax.xml.transform.TransformerException;
-
 import com.google.common.io.Resources;
 import com.google.inject.Inject;
+
+import org.killbill.billing.ErrorCode;
 import org.killbill.billing.catalog.StandaloneCatalog;
 import org.killbill.billing.catalog.VersionedCatalog;
 import org.killbill.billing.catalog.api.CatalogApiException;
-import org.killbill.billing.catalog.api.InvalidConfigException;
 import org.killbill.billing.platform.api.KillbillService.ServiceException;
 import org.killbill.clock.Clock;
 import org.killbill.xmlloader.UriAccessor;
-import org.killbill.xmlloader.ValidationException;
 import org.killbill.xmlloader.XMLLoader;
-import org.xml.sax.SAXException;
 
-public class VersionedCatalogLoader implements ICatalogLoader {
+public class VersionedCatalogLoader implements CatalogLoader {
     private static final Object PROTOCOL_FOR_FILE = "file";
     private final String XML_EXTENSION = ".xml";
     private final Clock clock;
@@ -55,7 +50,7 @@ public class VersionedCatalogLoader implements ICatalogLoader {
       * @see org.killbill.billing.catalog.io.ICatalogLoader#load(java.lang.String)
       */
     @Override
-    public VersionedCatalog load(final String uriString) throws ServiceException {
+    public VersionedCatalog load(final String uriString) throws CatalogApiException {
         try {
             List<URI> xmlURIs = null;
 
@@ -85,14 +80,13 @@ public class VersionedCatalogLoader implements ICatalogLoader {
                 final StandaloneCatalog catalog = XMLLoader.getObjectFromUri(u, StandaloneCatalog.class);
                 result.add(catalog);
             }
-
             return result;
         } catch (Exception e) {
-            throw new ServiceException("Problem encountered loading catalog", e);
+            throw new CatalogApiException(ErrorCode.CAT_INVALID_DEFAULT, "Problem encountered loading catalog ", e);
         }
     }
 
-    public VersionedCatalog load(final List<String> catalogXMLs) throws ServiceException {
+    public VersionedCatalog load(final List<String> catalogXMLs) throws CatalogApiException {
         final VersionedCatalog result = new VersionedCatalog(clock);
         final URI uri;
         try {
@@ -104,7 +98,7 @@ public class VersionedCatalogLoader implements ICatalogLoader {
             }
             return result;
         } catch (Exception e) {
-            throw new ServiceException("Problem encountered loading catalog", e);
+            throw new CatalogApiException(ErrorCode.CAT_INVALID_DEFAULT, "Problem encountered loading catalog ", e);
         }
     }
 

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
new file mode 100644
index 0000000..9109dde
--- /dev/null
+++ b/catalog/src/test/java/org/killbill/billing/catalog/caching/TestEhCacheCatalogCache.java
@@ -0,0 +1,123 @@
+/*
+ * Copyright 2014 Groupon, Inc
+ * Copyright 2014 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.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.URI;
+import java.net.URISyntaxException;
+
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.catalog.CatalogTestSuiteNoDB;
+import org.killbill.billing.catalog.DefaultProduct;
+import org.killbill.billing.catalog.VersionedCatalog;
+import org.killbill.billing.catalog.api.CatalogApiException;
+import org.killbill.billing.util.cache.Cachable.CacheType;
+import org.killbill.xmlloader.UriAccessor;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.io.CharStreams;
+import com.google.common.io.Resources;
+
+public class TestEhCacheCatalogCache extends CatalogTestSuiteNoDB {
+
+    private CatalogCache catalogCache;
+
+    @BeforeMethod(groups = "fast")
+    protected void beforeMethod() throws Exception {
+        this.catalogCache = new EhCacheCatalogCache(cacheControllerDispatcher.getCacheController(CacheType.TENANT_CATALOG), loader);
+        cacheControllerDispatcher.clearAll();
+
+    }
+
+    //
+    // Verify CatalogCache throws CatalogApiException when used in mono-tenant and catalog system property has not been set
+    //
+    @Test(groups = "fast", expectedExceptions = CatalogApiException.class)
+    public void testMissingDefaultCatalog() throws CatalogApiException {
+        Mockito.when(tenantInternalApi.getTenantCatalogs((Mockito.<InternalTenantContext>any()))).thenReturn(ImmutableList.<String>of());
+        catalogCache.getCatalog(internalCallContext);
+    }
+
+    //
+    // Verify CatalogCache returns default catalog when system property has been set (and CatalogCache has been initialized)
+    //
+    @Test(groups = "fast")
+    public void testDefaultCatalog() throws CatalogApiException {
+        catalogCache.loadDefaultCatalog(Resources.getResource("SpyCarBasic.xml").toExternalForm());
+        Mockito.when(tenantInternalApi.getTenantCatalogs((Mockito.<InternalTenantContext>any()))).thenReturn(ImmutableList.<String>of());
+        VersionedCatalog result = catalogCache.getCatalog(internalCallContext);
+        Assert.assertNotNull(result);
+        final DefaultProduct[] products = result.getProducts(clock.getUTCNow());
+        Assert.assertEquals(products.length, 3);
+    }
+
+    //
+    // Verify CatalogCache returns default catalog for the (non 0) tenant when its tenant catalog has not been uploaded
+    //
+    @Test(groups = "fast")
+    public void testMissingTenantCatalog() throws CatalogApiException, URISyntaxException, IOException {
+
+        catalogCache.loadDefaultCatalog(Resources.getResource("SpyCarBasic.xml").toExternalForm());
+
+        final InternalTenantContext tenantContext = Mockito.mock(InternalTenantContext.class);
+        Mockito.when(tenantContext.getTenantRecordId()).thenReturn(99L);
+
+        Mockito.when(tenantInternalApi.getTenantCatalogs((Mockito.<InternalTenantContext>any()))).thenReturn(ImmutableList.<String>of());
+        VersionedCatalog result = catalogCache.getCatalog(tenantContext);
+        Assert.assertNotNull(result);
+        final DefaultProduct[] products = result.getProducts(clock.getUTCNow());
+        Assert.assertEquals(products.length, 3);
+    }
+
+    //
+    // Verify CatalogCache returns per tenant catalog:
+    // 1. We first mock TenantInternalApi to return a different catalog than the default one
+    // 2. We then mock TenantInternalApi to throw RuntimeException which means catalog was cached and there was no additional call
+    //    to the TenantInternalApi api (otherwise test would fail with RuntimeException)
+    //
+    @Test(groups = "fast")
+    public void testExistingTenantCatalog() throws CatalogApiException, URISyntaxException, IOException {
+
+        catalogCache.loadDefaultCatalog(Resources.getResource("SpyCarBasic.xml").toExternalForm());
+
+        final InputStream inputCatalog = UriAccessor.accessUri(new URI(Resources.getResource("SpyCarAdvanced.xml").toExternalForm()));
+        final String catalogXML = CharStreams.toString(new InputStreamReader(inputCatalog, "UTF-8"));
+
+        final InternalTenantContext tenantContext = Mockito.mock(InternalTenantContext.class);
+        Mockito.when(tenantContext.getTenantRecordId()).thenReturn(156L);
+
+        Mockito.when(tenantInternalApi.getTenantCatalogs((Mockito.<InternalTenantContext>any()))).thenReturn(ImmutableList.<String>of(catalogXML));
+        VersionedCatalog result = catalogCache.getCatalog(tenantContext);
+        Assert.assertNotNull(result);
+        final DefaultProduct[] products = result.getProducts(clock.getUTCNow());
+        Assert.assertEquals(products.length, 6);
+
+        Mockito.when(tenantInternalApi.getTenantCatalogs(tenantContext)).thenThrow(RuntimeException.class);
+
+        VersionedCatalog result2 = catalogCache.getCatalog(tenantContext);
+        Assert.assertNotNull(result2);
+        final DefaultProduct[] products2 = result.getProducts(clock.getUTCNow());
+        Assert.assertEquals(products2.length, 6);
+    }
+}

diff --git a/catalog/src/test/java/org/killbill/billing/catalog/CatalogTestSuiteNoDB.java b/catalog/src/test/java/org/killbill/billing/catalog/CatalogTestSuiteNoDB.java
index c3d724f..d121d91 100644
--- a/catalog/src/test/java/org/killbill/billing/catalog/CatalogTestSuiteNoDB.java
+++ b/catalog/src/test/java/org/killbill/billing/catalog/CatalogTestSuiteNoDB.java
@@ -16,11 +16,13 @@
 
 package org.killbill.billing.catalog;
 
-import org.testng.annotations.BeforeClass;
-
 import org.killbill.billing.GuicyKillbillTestSuiteNoDB;
+import org.killbill.billing.catalog.caching.CatalogCache;
 import org.killbill.billing.catalog.glue.TestCatalogModuleNoDB;
 import org.killbill.billing.catalog.io.VersionedCatalogLoader;
+import org.killbill.billing.tenant.api.TenantInternalApi;
+import org.killbill.billing.util.cache.CacheControllerDispatcher;
+import org.testng.annotations.BeforeClass;
 
 import com.google.inject.Guice;
 import com.google.inject.Inject;
@@ -31,6 +33,12 @@ public abstract class CatalogTestSuiteNoDB extends GuicyKillbillTestSuiteNoDB {
     @Inject
     protected VersionedCatalogLoader loader;
 
+    @Inject
+    protected TenantInternalApi tenantInternalApi;
+
+    @Inject
+    protected CacheControllerDispatcher cacheControllerDispatcher;
+
     @BeforeClass(groups = "fast")
     protected void beforeClass() throws Exception {
         final Injector injector = Guice.createInjector(new TestCatalogModuleNoDB(configSource));

diff --git a/catalog/src/test/java/org/killbill/billing/catalog/io/TestVersionedCatalogLoader.java b/catalog/src/test/java/org/killbill/billing/catalog/io/TestVersionedCatalogLoader.java
index fb024c4..64d5075 100644
--- a/catalog/src/test/java/org/killbill/billing/catalog/io/TestVersionedCatalogLoader.java
+++ b/catalog/src/test/java/org/killbill/billing/catalog/io/TestVersionedCatalogLoader.java
@@ -33,6 +33,7 @@ import org.joda.time.DateTime;
 import org.killbill.billing.catalog.CatalogTestSuiteNoDB;
 import org.killbill.billing.catalog.StandaloneCatalog;
 import org.killbill.billing.catalog.VersionedCatalog;
+import org.killbill.billing.catalog.api.CatalogApiException;
 import org.killbill.billing.catalog.api.InvalidConfigException;
 import org.killbill.billing.platform.api.KillbillService.ServiceException;
 import org.testng.Assert;
@@ -115,7 +116,7 @@ public class TestVersionedCatalogLoader extends CatalogTestSuiteNoDB {
     }
 
     @Test(groups = "fast")
-    public void testLoad() throws IOException, SAXException, InvalidConfigException, JAXBException, TransformerException, URISyntaxException, ServiceException {
+    public void testLoad() throws IOException, SAXException, InvalidConfigException, JAXBException, TransformerException, URISyntaxException, CatalogApiException {
         final VersionedCatalog c = loader.load(Resources.getResource("versionedCatalog").toString());
         Assert.assertEquals(c.size(), 3);
         final Iterator<StandaloneCatalog> it = c.iterator();

diff --git a/catalog/src/test/java/org/killbill/billing/catalog/MockCatalogService.java b/catalog/src/test/java/org/killbill/billing/catalog/MockCatalogService.java
index f5e4dc5..17f65f1 100644
--- a/catalog/src/test/java/org/killbill/billing/catalog/MockCatalogService.java
+++ b/catalog/src/test/java/org/killbill/billing/catalog/MockCatalogService.java
@@ -19,13 +19,14 @@ package org.killbill.billing.catalog;
 import org.killbill.billing.callcontext.InternalTenantContext;
 import org.killbill.billing.catalog.api.Catalog;
 import org.killbill.billing.catalog.api.StaticCatalog;
+import org.killbill.billing.util.cache.CacheControllerDispatcher;
 
 public class MockCatalogService extends DefaultCatalogService {
 
     private final MockCatalog catalog;
 
-    public MockCatalogService(final MockCatalog catalog) {
-        super(null, null, null);
+    public MockCatalogService(final MockCatalog catalog, final CacheControllerDispatcher cacheControllerDispatcher) {
+        super(null, null, cacheControllerDispatcher);
         this.catalog = catalog;
     }
 

diff --git a/catalog/src/test/java/org/killbill/billing/catalog/TestCatalogService.java b/catalog/src/test/java/org/killbill/billing/catalog/TestCatalogService.java
index 95336cc..f84c056 100644
--- a/catalog/src/test/java/org/killbill/billing/catalog/TestCatalogService.java
+++ b/catalog/src/test/java/org/killbill/billing/catalog/TestCatalogService.java
@@ -21,6 +21,7 @@ package org.killbill.billing.catalog;
 import org.killbill.billing.catalog.api.CatalogApiException;
 import org.killbill.billing.catalog.io.VersionedCatalogLoader;
 import org.killbill.billing.platform.api.KillbillService.ServiceException;
+import org.killbill.billing.util.cache.CacheControllerDispatcher;
 import org.killbill.billing.util.config.CatalogConfig;
 import org.killbill.clock.DefaultClock;
 import org.testng.Assert;
@@ -36,7 +37,7 @@ public class TestCatalogService extends CatalogTestSuiteNoDB {
                 return "file:src/test/resources/versionedCatalog";
             }
 
-        }, null,  new VersionedCatalogLoader(new DefaultClock()));
+        }, new VersionedCatalogLoader(new DefaultClock()), cacheControllerDispatcher);
         service.loadCatalog();
         Assert.assertNotNull(service.getFullCatalog(internalCallContext));
         Assert.assertEquals(service.getFullCatalog(internalCallContext).getCatalogName(), "WeaponsHireSmall");
@@ -50,7 +51,7 @@ public class TestCatalogService extends CatalogTestSuiteNoDB {
                 return "file:src/test/resources/WeaponsHire.xml";
             }
 
-        }, null, new VersionedCatalogLoader(new DefaultClock()));
+        },  new VersionedCatalogLoader(new DefaultClock()), cacheControllerDispatcher);
         service.loadCatalog();
         Assert.assertNotNull(service.getFullCatalog(internalCallContext));
         Assert.assertEquals(service.getFullCatalog(internalCallContext).getCatalogName(), "Firearms");

diff --git a/subscription/src/test/java/org/killbill/billing/subscription/api/transfer/TestDefaultSubscriptionTransferApi.java b/subscription/src/test/java/org/killbill/billing/subscription/api/transfer/TestDefaultSubscriptionTransferApi.java
index d651c1b..9aed26d 100644
--- a/subscription/src/test/java/org/killbill/billing/subscription/api/transfer/TestDefaultSubscriptionTransferApi.java
+++ b/subscription/src/test/java/org/killbill/billing/subscription/api/transfer/TestDefaultSubscriptionTransferApi.java
@@ -62,7 +62,7 @@ public class TestDefaultSubscriptionTransferApi extends SubscriptionTestSuiteNoD
         super.beforeMethod();
         final NonEntityDao nonEntityDao = Mockito.mock(NonEntityDao.class);
         final SubscriptionDao dao = Mockito.mock(SubscriptionDao.class);
-        final CatalogService catalogService = new MockCatalogService(new MockCatalog());
+         final CatalogService catalogService = new MockCatalogService(new MockCatalog(), cacheControllerDispatcher);
         final SubscriptionBaseApiService apiService = Mockito.mock(SubscriptionBaseApiService.class);
         final SubscriptionBaseTimelineApi timelineApi = Mockito.mock(SubscriptionBaseTimelineApi.class);
         final InternalCallContextFactory internalCallContextFactory = new InternalCallContextFactory(clock, nonEntityDao, new CacheControllerDispatcher());

diff --git a/subscription/src/test/java/org/killbill/billing/subscription/SubscriptionTestSuiteNoDB.java b/subscription/src/test/java/org/killbill/billing/subscription/SubscriptionTestSuiteNoDB.java
index dbb011d..581397b 100644
--- a/subscription/src/test/java/org/killbill/billing/subscription/SubscriptionTestSuiteNoDB.java
+++ b/subscription/src/test/java/org/killbill/billing/subscription/SubscriptionTestSuiteNoDB.java
@@ -37,6 +37,7 @@ import org.killbill.billing.subscription.api.user.TestSubscriptionHelper;
 import org.killbill.billing.subscription.engine.dao.MockSubscriptionDaoMemory;
 import org.killbill.billing.subscription.engine.dao.SubscriptionDao;
 import org.killbill.billing.subscription.glue.TestDefaultSubscriptionModuleNoDB;
+import org.killbill.billing.util.cache.CacheControllerDispatcher;
 import org.killbill.billing.util.config.SubscriptionConfig;
 import org.killbill.clock.ClockMock;
 import org.mockito.Mockito;
@@ -90,10 +91,14 @@ public class SubscriptionTestSuiteNoDB extends GuicyKillbillTestSuiteNoDB {
     @Inject
     protected SubscriptionTestInitializer subscriptionTestInitializer;
 
+    @Inject
+    protected CacheControllerDispatcher cacheControllerDispatcher;
+
     protected Catalog catalog;
     protected AccountData accountData;
     protected SubscriptionBaseBundle bundle;
 
+
     @Override
     protected KillbillConfigSource getConfigSource() {
         return getConfigSource("/subscription.properties");

diff --git a/tenant/src/main/java/org/killbill/billing/tenant/api/DefaultTenantInternalApi.java b/tenant/src/main/java/org/killbill/billing/tenant/api/DefaultTenantInternalApi.java
index 50d1b7b..4d27d09 100644
--- a/tenant/src/main/java/org/killbill/billing/tenant/api/DefaultTenantInternalApi.java
+++ b/tenant/src/main/java/org/killbill/billing/tenant/api/DefaultTenantInternalApi.java
@@ -20,22 +20,24 @@ package org.killbill.billing.tenant.api;
 import java.util.List;
 
 import javax.inject.Inject;
+import javax.inject.Named;
 
 import org.killbill.billing.callcontext.InternalTenantContext;
 import org.killbill.billing.tenant.api.TenantKV.TenantKey;
 import org.killbill.billing.tenant.dao.TenantDao;
+import org.killbill.billing.tenant.glue.DefaultTenantModule;
 
 public class DefaultTenantInternalApi implements TenantInternalApi {
 
     private final TenantDao tenantDao;
 
     @Inject
-    public DefaultTenantInternalApi(final TenantDao tenantDao) {
+    public DefaultTenantInternalApi(@Named(DefaultTenantModule.NO_CACHING_TENANT) final TenantDao tenantDao) {
         this.tenantDao = tenantDao;
     }
 
     @Override
-    public List<String> getTenantCatalogs(final InternalTenantContext tenantContext) throws TenantApiException {
+    public List<String> getTenantCatalogs(final InternalTenantContext tenantContext) {
         return tenantDao.getTenantValueForKey(TenantKey.CATALOG.toString(), tenantContext);
     }
 }

diff --git a/tenant/src/main/java/org/killbill/billing/tenant/dao/NoCachingTenantDao.java b/tenant/src/main/java/org/killbill/billing/tenant/dao/NoCachingTenantDao.java
new file mode 100644
index 0000000..2d20bc0
--- /dev/null
+++ b/tenant/src/main/java/org/killbill/billing/tenant/dao/NoCachingTenantDao.java
@@ -0,0 +1,133 @@
+/*
+ * Copyright 2014 Groupon, Inc
+ * Copyright 2014 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.tenant.dao;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.killbill.billing.callcontext.InternalCallContext;
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.tenant.api.Tenant;
+import org.killbill.billing.tenant.api.TenantApiException;
+import org.killbill.billing.util.entity.Pagination;
+import org.killbill.billing.util.entity.dao.EntityDaoBase;
+import org.killbill.billing.util.entity.dao.EntitySqlDao;
+import org.killbill.billing.util.entity.dao.EntitySqlDaoTransactionWrapper;
+import org.killbill.billing.util.entity.dao.EntitySqlDaoTransactionalJdbiWrapper;
+import org.killbill.billing.util.entity.dao.EntitySqlDaoWrapperFactory;
+import org.killbill.clock.Clock;
+import org.skife.jdbi.v2.IDBI;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Collections2;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Inject;
+
+/**
+ * This is a special implementation of the TenantDao that does not rely on caching (CacheControllerDispatcher is not injected and passed
+ * to the EntitySqlDaoWrapperInvocationHandler. It only implements the set of operations that does not require caching:
+ * - Only getXXX methods where there is no Cache annotation on the SqlDao method
+ * - In addition excludes getById method, which are cached by EntitySqlDaoWrapperInvocationHandler
+ * <p/>
+ * <p/>
+ * It is used from the TenantInternalApi so that caching of catalog, overdue, ... can be done at a higher level (catalog module, overdue module)
+ * without causing guice dependency issues.
+ */
+public class NoCachingTenantDao extends EntityDaoBase<TenantModelDao, Tenant, TenantApiException> implements TenantDao {
+
+    @Inject
+    public NoCachingTenantDao(final IDBI dbi, final Clock clock) {
+        super(new EntitySqlDaoTransactionalJdbiWrapper(dbi, clock, null, null), TenantSqlDao.class);
+    }
+
+    @Override
+    public TenantModelDao getTenantByApiKey(final String key) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public List<String> getTenantValueForKey(final String key, final InternalTenantContext context) {
+        return transactionalSqlDao.execute(new EntitySqlDaoTransactionWrapper<List<String>>() {
+            @Override
+            public List<String> inTransaction(final EntitySqlDaoWrapperFactory<EntitySqlDao> entitySqlDaoWrapperFactory) throws Exception {
+                final List<TenantKVModelDao> tenantKV = entitySqlDaoWrapperFactory.become(TenantKVSqlDao.class).getTenantValueForKey(key, context);
+                return ImmutableList.copyOf(Collections2.transform(tenantKV, new Function<TenantKVModelDao, String>() {
+                    @Override
+                    public String apply(final TenantKVModelDao in) {
+                        return in.getTenantValue();
+                    }
+                }));
+            }
+        });
+    }
+
+    @Override
+    public void addTenantKeyValue(final String key, final String value, final InternalCallContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public void deleteTenantKey(final String key, final InternalCallContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public void create(final TenantModelDao entity, final InternalCallContext context) throws TenantApiException {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    protected TenantApiException generateAlreadyExistsException(final TenantModelDao entity, final InternalCallContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public Long getRecordId(final UUID id, final InternalTenantContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public TenantModelDao getByRecordId(final Long recordId, final InternalTenantContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public TenantModelDao getById(final UUID id, final InternalTenantContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public Pagination<TenantModelDao> getAll(final InternalTenantContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public Pagination<TenantModelDao> get(final Long offset, final Long limit, final InternalTenantContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public Long getCount(final InternalTenantContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+
+    @Override
+    public void test(final InternalTenantContext context) {
+        throw new IllegalStateException("Not implemented by NoCachingTenantDao");
+    }
+}

diff --git a/tenant/src/main/java/org/killbill/billing/tenant/glue/DefaultTenantModule.java b/tenant/src/main/java/org/killbill/billing/tenant/glue/DefaultTenantModule.java
index c2e842c..4df178b 100644
--- a/tenant/src/main/java/org/killbill/billing/tenant/glue/DefaultTenantModule.java
+++ b/tenant/src/main/java/org/killbill/billing/tenant/glue/DefaultTenantModule.java
@@ -27,11 +27,16 @@ import org.killbill.billing.tenant.api.TenantService;
 import org.killbill.billing.tenant.api.TenantUserApi;
 import org.killbill.billing.tenant.api.user.DefaultTenantUserApi;
 import org.killbill.billing.tenant.dao.DefaultTenantDao;
+import org.killbill.billing.tenant.dao.NoCachingTenantDao;
 import org.killbill.billing.tenant.dao.TenantDao;
 import org.killbill.billing.util.glue.KillBillModule;
 
+import com.google.inject.name.Names;
+
 public class DefaultTenantModule  extends KillBillModule implements TenantModule {
 
+    public static final String NO_CACHING_TENANT = "NoCachingTenant";
+
     public DefaultTenantModule(final KillbillConfigSource configSource) {
         super(configSource);
     }
@@ -41,6 +46,7 @@ public class DefaultTenantModule  extends KillBillModule implements TenantModule
 
     public void installTenantDao() {
         bind(TenantDao.class).to(DefaultTenantDao.class).asEagerSingleton();
+        bind(TenantDao.class).annotatedWith(Names.named(NO_CACHING_TENANT)).to(NoCachingTenantDao.class).asEagerSingleton();
     }
 
     public void installTenantUserApi() {

diff --git a/util/src/main/java/org/killbill/billing/util/cache/AccountRecordIdCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/AccountRecordIdCacheLoader.java
index 6c7e01e..7562ea9 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/AccountRecordIdCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/AccountRecordIdCacheLoader.java
@@ -32,9 +32,12 @@ import net.sf.ehcache.loader.CacheLoader;
 @Singleton
 public class AccountRecordIdCacheLoader extends BaseIdCacheLoader implements CacheLoader {
 
+    private final NonEntityDao nonEntityDao;
+
     @Inject
-    public AccountRecordIdCacheLoader(final IDBI dbi, final NonEntityDao nonEntityDao) {
-        super(dbi, nonEntityDao);
+    public AccountRecordIdCacheLoader(final NonEntityDao nonEntityDao) {
+        super();
+        this.nonEntityDao = nonEntityDao;
     }
 
     @Override

diff --git a/util/src/main/java/org/killbill/billing/util/cache/AuditLogCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/AuditLogCacheLoader.java
index 329bba5..bc0761d 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/AuditLogCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/AuditLogCacheLoader.java
@@ -35,7 +35,7 @@ public class AuditLogCacheLoader extends BaseCacheLoader implements CacheLoader 
 
     @Inject
     public AuditLogCacheLoader(final IDBI dbi, final NonEntityDao nonEntityDao) {
-        super(dbi, nonEntityDao);
+        super();
         this.auditSqlDao = dbi.onDemand(AuditSqlDao.class);
     }
 

diff --git a/util/src/main/java/org/killbill/billing/util/cache/AuditLogViaHistoryCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/AuditLogViaHistoryCacheLoader.java
index 8b1e76c..b819148 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/AuditLogViaHistoryCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/AuditLogViaHistoryCacheLoader.java
@@ -35,7 +35,7 @@ public class AuditLogViaHistoryCacheLoader extends BaseCacheLoader implements Ca
 
     @Inject
     public AuditLogViaHistoryCacheLoader(final IDBI dbi, final NonEntityDao nonEntityDao) {
-        super(dbi, nonEntityDao);
+        super();
         this.auditSqlDao = dbi.onDemand(AuditSqlDao.class);
     }
 

diff --git a/util/src/main/java/org/killbill/billing/util/cache/BaseCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/BaseCacheLoader.java
index 782cf0a..7aa7472 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/BaseCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/BaseCacheLoader.java
@@ -21,10 +21,7 @@ import java.util.Map;
 
 import javax.inject.Inject;
 
-import org.skife.jdbi.v2.IDBI;
-
 import org.killbill.billing.util.cache.Cachable.CacheType;
-import org.killbill.billing.util.dao.NonEntityDao;
 
 import net.sf.ehcache.CacheException;
 import net.sf.ehcache.Ehcache;
@@ -33,15 +30,10 @@ import net.sf.ehcache.loader.CacheLoader;
 
 public abstract class BaseCacheLoader implements CacheLoader {
 
-    protected final IDBI dbi;
-    protected final NonEntityDao nonEntityDao;
-
     private Status cacheLoaderStatus;
 
     @Inject
-    public BaseCacheLoader(final IDBI dbi, final NonEntityDao nonEntityDao) {
-        this.dbi = dbi;
-        this.nonEntityDao = nonEntityDao;
+    public BaseCacheLoader() {
         this.cacheLoaderStatus = Status.STATUS_UNINITIALISED;
     }
 

diff --git a/util/src/main/java/org/killbill/billing/util/cache/BaseIdCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/BaseIdCacheLoader.java
index 3ff9822..22ba37d 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/BaseIdCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/BaseIdCacheLoader.java
@@ -24,8 +24,8 @@ import org.skife.jdbi.v2.IDBI;
 
 public abstract class BaseIdCacheLoader extends BaseCacheLoader {
 
-    protected BaseIdCacheLoader(final IDBI dbi, final NonEntityDao nonEntityDao) {
-        super(dbi, nonEntityDao);
+    protected BaseIdCacheLoader() {
+        super();
     }
 
     @Override

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 6ede4f6..a3f3fd1 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
@@ -31,6 +31,7 @@ public @interface Cachable {
     public final String OBJECT_ID_CACHE_NAME = "object-id";
     public final String AUDIT_LOG_CACHE_NAME = "audit-log";
     public final String AUDIT_LOG_VIA_HISTORY_CACHE_NAME = "audit-log-via-history";
+    public final String TENANT_CATALOG_CACHE_NAME = "tenant-catalog";
 
     public CacheType value();
 
@@ -51,7 +52,10 @@ public @interface Cachable {
         AUDIT_LOG(AUDIT_LOG_CACHE_NAME, true),
 
         /* Mapping from object 'tableName::historyTableName::targetRecordId' -> matching objects 'Iterable<AuditLog>' */
-        AUDIT_LOG_VIA_HISTORY(AUDIT_LOG_VIA_HISTORY_CACHE_NAME, true);
+        AUDIT_LOG_VIA_HISTORY(AUDIT_LOG_VIA_HISTORY_CACHE_NAME, true),
+
+        /* Tenant catalog cache */
+        TENANT_CATALOG(TENANT_CATALOG_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 c8d7578..adea4d8 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
@@ -45,7 +45,8 @@ public class EhCacheCacheManagerProvider implements Provider<CacheManager> {
                                        final TenantRecordIdCacheLoader tenantRecordIdCacheLoader,
                                        final ObjectIdCacheLoader objectIdCacheLoader,
                                        final AuditLogCacheLoader auditLogCacheLoader,
-                                       final AuditLogViaHistoryCacheLoader auditLogViaHistoryCacheLoader) {
+                                       final AuditLogViaHistoryCacheLoader auditLogViaHistoryCacheLoader,
+                                       final TenantCatalogCacheLoader tenantCatalogCacheLoader) {
         this.cacheConfig = cacheConfig;
         cacheLoaders.add(recordIdCacheLoader);
         cacheLoaders.add(accountRecordIdCacheLoader);
@@ -53,6 +54,7 @@ public class EhCacheCacheManagerProvider implements Provider<CacheManager> {
         cacheLoaders.add(objectIdCacheLoader);
         cacheLoaders.add(auditLogCacheLoader);
         cacheLoaders.add(auditLogViaHistoryCacheLoader);
+        cacheLoaders.add(tenantCatalogCacheLoader);
     }
 
     @Override

diff --git a/util/src/main/java/org/killbill/billing/util/cache/ObjectIdCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/ObjectIdCacheLoader.java
index f770184..b4010bf 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/ObjectIdCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/ObjectIdCacheLoader.java
@@ -31,9 +31,12 @@ import net.sf.ehcache.loader.CacheLoader;
 @Singleton
 public class ObjectIdCacheLoader extends BaseIdCacheLoader implements CacheLoader {
 
+    private final NonEntityDao nonEntityDao;
+
     @Inject
-    public ObjectIdCacheLoader(final IDBI dbi, final NonEntityDao nonEntityDao) {
-        super(dbi, nonEntityDao);
+    public ObjectIdCacheLoader(final NonEntityDao nonEntityDao) {
+        super();
+        this.nonEntityDao = nonEntityDao;
     }
 
     @Override

diff --git a/util/src/main/java/org/killbill/billing/util/cache/RecordIdCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/RecordIdCacheLoader.java
index 8398911..6c78335 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/RecordIdCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/RecordIdCacheLoader.java
@@ -27,15 +27,17 @@ import org.killbill.billing.ObjectType;
 import org.killbill.billing.util.cache.Cachable.CacheType;
 import org.killbill.billing.util.dao.NonEntityDao;
 
-import net.sf.ehcache.CacheException;
 import net.sf.ehcache.loader.CacheLoader;
 
 @Singleton
 public class RecordIdCacheLoader extends BaseIdCacheLoader implements CacheLoader {
 
+    private final NonEntityDao nonEntityDao;
+
     @Inject
     public RecordIdCacheLoader(final IDBI dbi, final NonEntityDao nonEntityDao) {
-        super(dbi, nonEntityDao);
+        super();
+        this.nonEntityDao = nonEntityDao;
     }
 
     @Override

diff --git a/util/src/main/java/org/killbill/billing/util/cache/TenantCatalogCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/TenantCatalogCacheLoader.java
new file mode 100644
index 0000000..63017dc
--- /dev/null
+++ b/util/src/main/java/org/killbill/billing/util/cache/TenantCatalogCacheLoader.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright 2014 Groupon, Inc
+ * Copyright 2014 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.tenant.api.TenantInternalApi;
+import org.killbill.billing.util.cache.Cachable.CacheType;
+
+@Singleton
+public class TenantCatalogCacheLoader extends BaseCacheLoader {
+
+    private final TenantInternalApi tenantApi;
+
+    @Inject
+    public TenantCatalogCacheLoader(final TenantInternalApi tenantApi) {
+        super();
+        this.tenantApi = tenantApi;
+    }
+
+    @Override
+    public CacheType getCacheType() {
+        return CacheType.TENANT_CATALOG;
+    }
+
+    @Override
+    public Object load(final Object key, final Object argument) {
+        checkCacheLoaderStatus();
+
+        if (!(key instanceof InternalTenantContext)) {
+            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 InternalTenantContext internalTenantContext = (InternalTenantContext) key;
+        final CacheLoaderArgument cacheLoaderArgument = (CacheLoaderArgument) argument;
+
+        if (cacheLoaderArgument.getArgs() == null || !(cacheLoaderArgument.getArgs()[0] instanceof LoaderCallback)) {
+            throw new IllegalArgumentException("Missing LoaderCallback from the arguments ");
+        }
+
+        final LoaderCallback callback = (LoaderCallback) cacheLoaderArgument.getArgs()[0];
+
+        final List<String> catalogXMLs = tenantApi.getTenantCatalogs(internalTenantContext);
+        if (catalogXMLs.isEmpty()) {
+            return null;
+        }
+        try {
+            return callback.loadCatalog(catalogXMLs);
+        } catch (CatalogApiException e) {
+            throw new IllegalStateException(String.format("Failed to de-serialize catalog for tenant %s : %s",
+                                                          internalTenantContext.getTenantRecordId(), e.getMessage()), e);
+        }
+    }
+
+    public interface LoaderCallback {
+        public Object loadCatalog(final List<String> catalogXMLs) throws CatalogApiException;
+    }
+}

diff --git a/util/src/main/java/org/killbill/billing/util/cache/TenantRecordIdCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/TenantRecordIdCacheLoader.java
index 008fc52..fe3d2f4 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/TenantRecordIdCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/TenantRecordIdCacheLoader.java
@@ -32,9 +32,12 @@ import net.sf.ehcache.loader.CacheLoader;
 @Singleton
 public class TenantRecordIdCacheLoader extends BaseIdCacheLoader implements CacheLoader {
 
+    private final NonEntityDao nonEntityDao;
+
     @Inject
-    public TenantRecordIdCacheLoader(final IDBI dbi, final NonEntityDao nonEntityDao) {
-        super(dbi, nonEntityDao);
+    public TenantRecordIdCacheLoader(final NonEntityDao nonEntityDao) {
+        super();
+        this.nonEntityDao = nonEntityDao;
     }
 
     @Override

diff --git a/util/src/main/java/org/killbill/billing/util/entity/dao/EntitySqlDaoWrapperInvocationHandler.java b/util/src/main/java/org/killbill/billing/util/entity/dao/EntitySqlDaoWrapperInvocationHandler.java
index 1432259..89a2c1d 100644
--- a/util/src/main/java/org/killbill/billing/util/entity/dao/EntitySqlDaoWrapperInvocationHandler.java
+++ b/util/src/main/java/org/killbill/billing/util/entity/dao/EntitySqlDaoWrapperInvocationHandler.java
@@ -29,6 +29,8 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
+import javax.annotation.Nullable;
+
 import org.killbill.billing.ObjectType;
 import org.killbill.billing.callcontext.InternalCallContext;
 import org.killbill.billing.callcontext.InternalTenantContext;
@@ -86,7 +88,12 @@ public class EntitySqlDaoWrapperInvocationHandler<S extends EntitySqlDao<M, E>, 
     private final NonEntityDao nonEntityDao;
     private final Profiling prof;
 
-    public EntitySqlDaoWrapperInvocationHandler(final Class<S> sqlDaoClass, final S sqlDao, final Clock clock, final CacheControllerDispatcher cacheControllerDispatcher, final NonEntityDao nonEntityDao) {
+    public EntitySqlDaoWrapperInvocationHandler(final Class<S> sqlDaoClass,
+                                                final S sqlDao,
+                                                final Clock clock,
+                                                // Special DAO that don't require caching can invoke EntitySqlDaoWrapperInvocationHandler with no caching (e.g NoCachingTenantDao)
+                                                @Nullable final CacheControllerDispatcher cacheControllerDispatcher,
+                                                @Nullable final NonEntityDao nonEntityDao) {
         this.sqlDaoClass = sqlDaoClass;
         this.sqlDao = sqlDao;
         this.clock = clock;

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

diff --git a/util/src/test/java/org/killbill/billing/util/glue/TestUtilModule.java b/util/src/test/java/org/killbill/billing/util/glue/TestUtilModule.java
index 08c2b2e..9dee2b4 100644
--- a/util/src/test/java/org/killbill/billing/util/glue/TestUtilModule.java
+++ b/util/src/test/java/org/killbill/billing/util/glue/TestUtilModule.java
@@ -18,6 +18,7 @@
 
 package org.killbill.billing.util.glue;
 
+import org.killbill.billing.mock.glue.MockTenantModule;
 import org.killbill.billing.platform.api.KillbillConfigSource;
 import org.killbill.billing.subscription.api.timeline.SubscriptionBaseTimelineApi;
 import org.mockito.Mockito;
@@ -37,7 +38,7 @@ public class TestUtilModule extends KillBillModule {
     protected void configure() {
         //install(new CallContextModule());
         install(new CacheModule(configSource));
-
+        install(new MockTenantModule(configSource));
         installHack();
     }
 }