killbill-uncached

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 071f8ed..6480aea 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
@@ -21,6 +21,7 @@ import java.util.List;
 import java.util.Locale;
 
 import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.tenant.api.TenantKV.TenantKey;
 
 public interface TenantInternalApi {
 
@@ -28,9 +29,11 @@ public interface TenantInternalApi {
         public void invalidateCache(InternalTenantContext tenantContext);
     }
 
-    public List<String> getTenantCatalogs(InternalTenantContext tenantContext, CacheInvalidationCallback cacheInvalidationCallback);
+    public void initializeCacheInvalidationCallback(final TenantKey key, final CacheInvalidationCallback cacheInvalidationCallback);
 
-    public String getTenantOverdueConfig(InternalTenantContext tenantContext, CacheInvalidationCallback cacheInvalidationCallback);
+    public List<String> getTenantCatalogs(InternalTenantContext tenantContext);
+
+    public String getTenantOverdueConfig(InternalTenantContext tenantContext);
 
     public String getInvoiceTemplate(Locale locale, InternalTenantContext tenantContext);
 

diff --git a/catalog/src/main/java/org/killbill/billing/catalog/caching/CatalogCacheInvalidationCallback.java b/catalog/src/main/java/org/killbill/billing/catalog/caching/CatalogCacheInvalidationCallback.java
new file mode 100644
index 0000000..eef2e24
--- /dev/null
+++ b/catalog/src/main/java/org/killbill/billing/catalog/caching/CatalogCacheInvalidationCallback.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2015 Groupon, Inc
+ * Copyright 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 javax.inject.Inject;
+
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.tenant.api.TenantInternalApi.CacheInvalidationCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CatalogCacheInvalidationCallback implements CacheInvalidationCallback {
+
+    private final Logger logger = LoggerFactory.getLogger(CatalogCacheInvalidationCallback.class);
+
+    private final CatalogCache catalogCache;
+
+    @Inject
+    public CatalogCacheInvalidationCallback(final CatalogCache catalogCache) {
+        this.catalogCache = catalogCache;
+    }
+
+    @Override
+    public void invalidateCache(final InternalTenantContext tenantContext) {
+        logger.info("Invalidate catalog cache for tenant " + tenantContext.getTenantRecordId());
+        catalogCache.clearCatalog(tenantContext);
+    }
+}

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 256b8e2..acfac0c 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
@@ -33,9 +33,13 @@ import org.killbill.billing.util.cache.CacheControllerDispatcher;
 import org.killbill.billing.util.cache.CacheLoaderArgument;
 import org.killbill.billing.util.cache.TenantCatalogCacheLoader.LoaderCallback;
 import org.killbill.billing.util.callcontext.InternalCallContextFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class EhCacheCatalogCache implements CatalogCache {
 
+    private final Logger logger = LoggerFactory.getLogger(EhCacheCatalogCache.class);
+
     private final CacheController cacheController;
     private final VersionedCatalogLoader loader;
     private final CacheLoaderArgument cacheLoaderArgument;
@@ -89,11 +93,6 @@ public class EhCacheCatalogCache implements CatalogCache {
             public Object loadCatalog(final List<String> catalogXMLs) throws CatalogApiException {
                 return loader.load(catalogXMLs);
             }
-
-            @Override
-            public void invalidateCache(final InternalTenantContext tenantContext) {
-                parentCache.clearCatalog(tenantContext);
-            }
         };
         final Object[] args = new Object[1];
         args[0]= loaderCallback;

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 0497e9d..053122c 100644
--- a/catalog/src/main/java/org/killbill/billing/catalog/DefaultCatalogService.java
+++ b/catalog/src/main/java/org/killbill/billing/catalog/DefaultCatalogService.java
@@ -18,15 +18,22 @@
 
 package org.killbill.billing.catalog;
 
+import javax.inject.Named;
+
 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.CatalogCacheInvalidationCallback;
+import org.killbill.billing.catalog.glue.CatalogModule;
 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.TenantInternalApi;
+import org.killbill.billing.tenant.api.TenantInternalApi.CacheInvalidationCallback;
+import org.killbill.billing.tenant.api.TenantKV.TenantKey;
 import org.killbill.billing.util.config.CatalogConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,14 +48,21 @@ public class DefaultCatalogService implements KillbillService, CatalogService {
     private final CatalogConfig config;
     private boolean isInitialized;
 
+    private final TenantInternalApi tenantInternalApi;
+
     private final CatalogCache catalogCache;
+    private final CacheInvalidationCallback cacheInvalidationCallback;
 
     @Inject
     public DefaultCatalogService(final CatalogConfig config,
-                                 final CatalogCache catalogCache) {
+                                 final TenantInternalApi tenantInternalApi,
+                                 final CatalogCache catalogCache,
+                                 @Named(CatalogModule.CATALOG_INVALIDATION_CALLBACK) final CacheInvalidationCallback cacheInvalidationCallback) {
         this.config = config;
-        this.isInitialized = false;
         this.catalogCache = catalogCache;
+        this.cacheInvalidationCallback = cacheInvalidationCallback;
+        this.tenantInternalApi = tenantInternalApi;
+        this.isInitialized = false;
     }
 
     @LifecycleHandlerType(LifecycleLevel.LOAD_CATALOG)
@@ -67,7 +81,12 @@ public class DefaultCatalogService implements KillbillService, CatalogService {
         }
     }
 
-    @Override
+    @LifecycleHandlerType(LifecycleLevel.INIT_SERVICE)
+    public synchronized void initialize() throws ServiceException {
+        tenantInternalApi.initializeCacheInvalidationCallback(TenantKey.CATALOG, cacheInvalidationCallback);
+    }
+
+        @Override
     public String getName() {
         return CATALOG_SERVICE_NAME;
     }

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 2198ce6..7118e5f 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
@@ -23,16 +23,22 @@ 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.caching.CatalogCache;
+import org.killbill.billing.catalog.caching.CatalogCacheInvalidationCallback;
 import org.killbill.billing.catalog.caching.EhCacheCatalogCache;
 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.tenant.api.TenantInternalApi.CacheInvalidationCallback;
 import org.killbill.billing.util.config.CatalogConfig;
 import org.killbill.billing.util.glue.KillBillModule;
 import org.skife.config.ConfigurationObjectFactory;
 
+import com.google.inject.name.Names;
+
 public class CatalogModule extends KillBillModule {
 
+    public static final String CATALOG_INVALIDATION_CALLBACK = "CatalogInvalidationCallback";
+
     public CatalogModule(final KillbillConfigSource configSource) {
         super(configSource);
     }
@@ -51,8 +57,9 @@ public class CatalogModule extends KillBillModule {
         bind(CatalogUserApi.class).to(DefaultCatalogUserApi.class).asEagerSingleton();
     }
 
-    public void installOverdueConfigCache() {
+    public void installCatalogConfigCache() {
         bind(CatalogCache.class).to(EhCacheCatalogCache.class).asEagerSingleton();
+        bind(CacheInvalidationCallback.class).annotatedWith(Names.named(CATALOG_INVALIDATION_CALLBACK)).to(CatalogCacheInvalidationCallback.class).asEagerSingleton();
     }
 
     @Override
@@ -60,6 +67,6 @@ public class CatalogModule extends KillBillModule {
         installConfig();
         installCatalog();
         installCatalogUserApi();
-        installOverdueConfigCache();
+        installCatalogConfigCache();
     }
 }

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 5e0706e..336f691 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
@@ -57,7 +57,7 @@ public class TestEhCacheCatalogCache extends CatalogTestSuiteNoDB {
         final InternalTenantContext tenantContext = Mockito.mock(InternalTenantContext.class);
         Mockito.when(tenantContext.getTenantRecordId()).thenReturn(0L);
         catalogCache.loadDefaultCatalog(null);
-        Mockito.when(tenantInternalApi.getTenantCatalogs(tenantContext, cacheInvalidationCallback)).thenReturn(ImmutableList.<String>of());
+        Mockito.when(tenantInternalApi.getTenantCatalogs(tenantContext)).thenReturn(ImmutableList.<String>of());
         catalogCache.getCatalog(internalCallContext);
     }
 
@@ -71,7 +71,7 @@ public class TestEhCacheCatalogCache extends CatalogTestSuiteNoDB {
         Mockito.when(tenantContext.getTenantRecordId()).thenReturn(0L);
 
         catalogCache.loadDefaultCatalog(Resources.getResource("SpyCarBasic.xml").toExternalForm());
-        Mockito.when(tenantInternalApi.getTenantCatalogs(tenantContext, cacheInvalidationCallback)).thenReturn(ImmutableList.<String>of());
+        Mockito.when(tenantInternalApi.getTenantCatalogs(tenantContext)).thenReturn(ImmutableList.<String>of());
         VersionedCatalog result = catalogCache.getCatalog(internalCallContext);
         Assert.assertNotNull(result);
         final DefaultProduct[] products = result.getProducts(clock.getUTCNow());
@@ -89,7 +89,7 @@ public class TestEhCacheCatalogCache extends CatalogTestSuiteNoDB {
         final InternalTenantContext tenantContext = Mockito.mock(InternalTenantContext.class);
         Mockito.when(tenantContext.getTenantRecordId()).thenReturn(99L);
 
-        Mockito.when(tenantInternalApi.getTenantCatalogs(Mockito.any(InternalTenantContext.class), Mockito.any(CacheInvalidationCallback.class))).thenReturn(ImmutableList.<String>of());
+        Mockito.when(tenantInternalApi.getTenantCatalogs(Mockito.any(InternalTenantContext.class))).thenReturn(ImmutableList.<String>of());
         VersionedCatalog result = catalogCache.getCatalog(tenantContext);
         Assert.assertNotNull(result);
         final DefaultProduct[] products = result.getProducts(clock.getUTCNow());
@@ -113,13 +113,13 @@ public class TestEhCacheCatalogCache extends CatalogTestSuiteNoDB {
         final InternalTenantContext tenantContext = Mockito.mock(InternalTenantContext.class);
         Mockito.when(tenantContext.getTenantRecordId()).thenReturn(156L);
 
-        Mockito.when(tenantInternalApi.getTenantCatalogs(Mockito.any(InternalTenantContext.class), Mockito.any(CacheInvalidationCallback.class))).thenReturn(ImmutableList.<String>of(catalogXML));
+        Mockito.when(tenantInternalApi.getTenantCatalogs(Mockito.any(InternalTenantContext.class))).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, cacheInvalidationCallback)).thenThrow(RuntimeException.class);
+        Mockito.when(tenantInternalApi.getTenantCatalogs(tenantContext)).thenThrow(RuntimeException.class);
 
         VersionedCatalog result2 = catalogCache.getCatalog(tenantContext);
         Assert.assertNotNull(result2);

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 00d82b5..9b84483 100644
--- a/catalog/src/test/java/org/killbill/billing/catalog/CatalogTestSuiteNoDB.java
+++ b/catalog/src/test/java/org/killbill/billing/catalog/CatalogTestSuiteNoDB.java
@@ -18,6 +18,7 @@ package org.killbill.billing.catalog;
 
 import org.killbill.billing.GuicyKillbillTestSuiteNoDB;
 import org.killbill.billing.catalog.caching.CatalogCache;
+import org.killbill.billing.catalog.caching.CatalogCacheInvalidationCallback;
 import org.killbill.billing.catalog.glue.TestCatalogModuleNoDB;
 import org.killbill.billing.catalog.io.VersionedCatalogLoader;
 import org.killbill.billing.tenant.api.TenantInternalApi;
@@ -42,6 +43,9 @@ public abstract class CatalogTestSuiteNoDB extends GuicyKillbillTestSuiteNoDB {
     @Inject
     protected CatalogCache catalogCache;
 
+    @Inject
+    protected CatalogCacheInvalidationCallback cacheInvalidationCallback;
+
     @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/MockCatalogService.java b/catalog/src/test/java/org/killbill/billing/catalog/MockCatalogService.java
index 59a632c..41b66fa 100644
--- a/catalog/src/test/java/org/killbill/billing/catalog/MockCatalogService.java
+++ b/catalog/src/test/java/org/killbill/billing/catalog/MockCatalogService.java
@@ -26,7 +26,7 @@ public class MockCatalogService extends DefaultCatalogService {
     private final MockCatalog catalog;
 
     public MockCatalogService(final MockCatalog catalog, final CacheControllerDispatcher cacheControllerDispatcher) {
-        super(null, null);
+        super(null, null, null, null);
         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 47e7b7d..08a434b 100644
--- a/catalog/src/test/java/org/killbill/billing/catalog/TestCatalogService.java
+++ b/catalog/src/test/java/org/killbill/billing/catalog/TestCatalogService.java
@@ -36,7 +36,7 @@ public class TestCatalogService extends CatalogTestSuiteNoDB {
                 return "file:src/test/resources/versionedCatalog";
             }
 
-        }, catalogCache);
+        }, tenantInternalApi, catalogCache, cacheInvalidationCallback);
         service.loadCatalog();
         Assert.assertNotNull(service.getFullCatalog(internalCallContext));
         Assert.assertEquals(service.getFullCatalog(internalCallContext).getCatalogName(), "WeaponsHireSmall");
@@ -50,7 +50,7 @@ public class TestCatalogService extends CatalogTestSuiteNoDB {
                 return "file:src/test/resources/WeaponsHire.xml";
             }
 
-        },  catalogCache);
+        },  tenantInternalApi, catalogCache, cacheInvalidationCallback);
         service.loadCatalog();
         Assert.assertNotNull(service.getFullCatalog(internalCallContext));
         Assert.assertEquals(service.getFullCatalog(internalCallContext).getCatalogName(), "Firearms");

diff --git a/overdue/src/main/java/org/killbill/billing/overdue/caching/EhCacheOverdueConfigCache.java b/overdue/src/main/java/org/killbill/billing/overdue/caching/EhCacheOverdueConfigCache.java
index 5c7542d..dc998c9 100644
--- a/overdue/src/main/java/org/killbill/billing/overdue/caching/EhCacheOverdueConfigCache.java
+++ b/overdue/src/main/java/org/killbill/billing/overdue/caching/EhCacheOverdueConfigCache.java
@@ -27,7 +27,6 @@ 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.api.CatalogApiException;
 import org.killbill.billing.overdue.api.OverdueApiException;
 import org.killbill.billing.overdue.api.OverdueConfig;
 import org.killbill.billing.overdue.config.DefaultOverdueConfig;
@@ -53,7 +52,7 @@ public class EhCacheOverdueConfigCache implements OverdueConfigCache {
     @Inject
     public EhCacheOverdueConfigCache(final CacheControllerDispatcher cacheControllerDispatcher) {
         this.cacheController = cacheControllerDispatcher.getCacheController(CacheType.TENANT_OVERDUE_CONFIG);
-        this.cacheLoaderArgument = initializeCacheLoaderArgument(this);
+        this.cacheLoaderArgument = initializeCacheLoaderArgument();
     }
 
     @Override
@@ -107,7 +106,7 @@ public class EhCacheOverdueConfigCache implements OverdueConfigCache {
         cacheController.remove(tenantContext);
     }
 
-    private CacheLoaderArgument initializeCacheLoaderArgument(final EhCacheOverdueConfigCache parentCache) {
+    private CacheLoaderArgument initializeCacheLoaderArgument() {
         final LoaderCallback loaderCallback = new LoaderCallback() {
             @Override
             public Object loadCatalog(final String catalogXMLs) throws OverdueApiException {
@@ -121,11 +120,6 @@ public class EhCacheOverdueConfigCache implements OverdueConfigCache {
                     throw new OverdueApiException(ErrorCode.OVERDUE_INVALID_FOR_TENANT, "Problem encountered loading overdue config ", e);
                 }
             }
-
-            @Override
-            public void invalidateCache(final InternalTenantContext tenantContext) {
-                parentCache.clearOverdueConfig(tenantContext);
-            }
         };
         final Object[] args = new Object[1];
         args[0] = loaderCallback;

diff --git a/overdue/src/main/java/org/killbill/billing/overdue/caching/OverdueCacheInvalidationCallback.java b/overdue/src/main/java/org/killbill/billing/overdue/caching/OverdueCacheInvalidationCallback.java
new file mode 100644
index 0000000..c16ae9f
--- /dev/null
+++ b/overdue/src/main/java/org/killbill/billing/overdue/caching/OverdueCacheInvalidationCallback.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2015 Groupon, Inc
+ * Copyright 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.overdue.caching;
+
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.tenant.api.TenantInternalApi.CacheInvalidationCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+
+public class OverdueCacheInvalidationCallback implements CacheInvalidationCallback {
+
+    private static final Logger logger = LoggerFactory.getLogger(OverdueCacheInvalidationCallback.class);
+
+    private final OverdueConfigCache overdueConfigCache;
+
+    @Inject
+    public OverdueCacheInvalidationCallback(final OverdueConfigCache overdueConfigCache) {
+        this.overdueConfigCache = overdueConfigCache;
+    }
+
+    @Override
+    public void invalidateCache(final InternalTenantContext tenantContext) {
+        logger.info("Invalidate overdue cache for tenant " + tenantContext.getTenantRecordId());
+        overdueConfigCache.clearOverdueConfig(tenantContext);
+    }
+}

diff --git a/overdue/src/main/java/org/killbill/billing/overdue/glue/DefaultOverdueModule.java b/overdue/src/main/java/org/killbill/billing/overdue/glue/DefaultOverdueModule.java
index c79179c..9c613e5 100644
--- a/overdue/src/main/java/org/killbill/billing/overdue/glue/DefaultOverdueModule.java
+++ b/overdue/src/main/java/org/killbill/billing/overdue/glue/DefaultOverdueModule.java
@@ -29,6 +29,7 @@ import org.killbill.billing.overdue.applicator.OverdueEmailGenerator;
 import org.killbill.billing.overdue.applicator.formatters.DefaultOverdueEmailFormatterFactory;
 import org.killbill.billing.overdue.applicator.formatters.OverdueEmailFormatterFactory;
 import org.killbill.billing.overdue.caching.EhCacheOverdueConfigCache;
+import org.killbill.billing.overdue.caching.OverdueCacheInvalidationCallback;
 import org.killbill.billing.overdue.caching.OverdueConfigCache;
 import org.killbill.billing.overdue.listener.OverdueListener;
 import org.killbill.billing.overdue.notification.OverdueAsyncBusNotifier;
@@ -40,6 +41,7 @@ import org.killbill.billing.overdue.notification.OverduePoster;
 import org.killbill.billing.overdue.service.DefaultOverdueService;
 import org.killbill.billing.overdue.wrapper.OverdueWrapperFactory;
 import org.killbill.billing.platform.api.KillbillConfigSource;
+import org.killbill.billing.tenant.api.TenantInternalApi.CacheInvalidationCallback;
 import org.killbill.billing.util.glue.KillBillModule;
 import org.skife.config.ConfigurationObjectFactory;
 
@@ -47,6 +49,7 @@ import com.google.inject.name.Names;
 
 public class DefaultOverdueModule extends KillBillModule implements OverdueModule {
 
+    public static final String OVERDUE_INVALIDATION_CALLBACK = "overdueInvalidationCallback";
     public static final String OVERDUE_NOTIFIER_CHECK_NAMED = "overdueNotifierCheck";
     public static final String OVERDUE_NOTIFIER_ASYNC_BUS_NAMED = "overdueNotifierAsyncBus";
 
@@ -98,5 +101,6 @@ public class DefaultOverdueModule extends KillBillModule implements OverdueModul
 
     public void installOverdueConfigCache() {
         bind(OverdueConfigCache.class).to(EhCacheOverdueConfigCache.class).asEagerSingleton();
+        bind(CacheInvalidationCallback.class).annotatedWith(Names.named(OVERDUE_INVALIDATION_CALLBACK)).to(OverdueCacheInvalidationCallback.class).asEagerSingleton();
     }
 }

diff --git a/overdue/src/main/java/org/killbill/billing/overdue/service/DefaultOverdueService.java b/overdue/src/main/java/org/killbill/billing/overdue/service/DefaultOverdueService.java
index d7f37b7..011eb4f 100644
--- a/overdue/src/main/java/org/killbill/billing/overdue/service/DefaultOverdueService.java
+++ b/overdue/src/main/java/org/killbill/billing/overdue/service/DefaultOverdueService.java
@@ -18,9 +18,6 @@
 
 package org.killbill.billing.overdue.service;
 
-import java.net.URI;
-import java.net.URISyntaxException;
-
 import javax.inject.Named;
 
 import org.killbill.billing.callcontext.InternalTenantContext;
@@ -29,18 +26,16 @@ import org.killbill.billing.overdue.OverdueProperties;
 import org.killbill.billing.overdue.OverdueService;
 import org.killbill.billing.overdue.api.OverdueApiException;
 import org.killbill.billing.overdue.api.OverdueConfig;
-import org.killbill.billing.overdue.caching.EhCacheOverdueConfigCache;
 import org.killbill.billing.overdue.caching.OverdueConfigCache;
-import org.killbill.billing.overdue.config.DefaultOverdueConfig;
 import org.killbill.billing.overdue.glue.DefaultOverdueModule;
 import org.killbill.billing.overdue.listener.OverdueListener;
 import org.killbill.billing.overdue.notification.OverdueNotifier;
 import org.killbill.billing.platform.api.LifecycleHandlerType;
 import org.killbill.billing.platform.api.LifecycleHandlerType.LifecycleLevel;
-import org.killbill.billing.util.cache.Cachable.CacheType;
-import org.killbill.billing.util.cache.CacheControllerDispatcher;
+import org.killbill.billing.tenant.api.TenantInternalApi;
+import org.killbill.billing.tenant.api.TenantInternalApi.CacheInvalidationCallback;
+import org.killbill.billing.tenant.api.TenantKV.TenantKey;
 import org.killbill.bus.api.PersistentBus.EventBusException;
-import org.killbill.xmlloader.XMLLoader;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -59,8 +54,9 @@ public class DefaultOverdueService implements OverdueService {
     private final OverdueListener listener;
 
     private final OverdueConfigCache overdueConfigCache;
+    private final CacheInvalidationCallback overdueCacheInvalidationCallback;
 
-    private DefaultOverdueConfig overdueConfig;
+    private final TenantInternalApi tenantInternalApi;
     private boolean isConfigLoaded;
 
     @Inject
@@ -69,7 +65,9 @@ public class DefaultOverdueService implements OverdueService {
                                  @Named(DefaultOverdueModule.OVERDUE_NOTIFIER_ASYNC_BUS_NAMED) final OverdueNotifier asyncNotifier,
                                  final BusService busService,
                                  final OverdueListener listener,
-                                 final OverdueConfigCache overdueConfigCache) {
+                                 final OverdueConfigCache overdueConfigCache,
+                                 @Named(DefaultOverdueModule.OVERDUE_INVALIDATION_CALLBACK) final CacheInvalidationCallback overdueCacheInvalidationCallback,
+                                 final TenantInternalApi tenantInternalApi) {
         this.properties = properties;
         this.checkNotifier = checkNotifier;
         this.asyncNotifier = asyncNotifier;
@@ -77,6 +75,8 @@ public class DefaultOverdueService implements OverdueService {
         this.listener = listener;
         this.isConfigLoaded = false;
         this.overdueConfigCache = overdueConfigCache;
+        this.overdueCacheInvalidationCallback = overdueCacheInvalidationCallback;
+        this.tenantInternalApi = tenantInternalApi;
     }
 
     @Override
@@ -102,6 +102,7 @@ public class DefaultOverdueService implements OverdueService {
         registerForBus();
         checkNotifier.initialize();
         asyncNotifier.initialize();
+        tenantInternalApi.initializeCacheInvalidationCallback(TenantKey.OVERDUE_CONFIG, overdueCacheInvalidationCallback);
     }
 
     private void registerForBus() {

diff --git a/overdue/src/test/java/org/killbill/billing/overdue/glue/TestOverdueModule.java b/overdue/src/test/java/org/killbill/billing/overdue/glue/TestOverdueModule.java
index 0284e6d..d858aa1 100644
--- a/overdue/src/test/java/org/killbill/billing/overdue/glue/TestOverdueModule.java
+++ b/overdue/src/test/java/org/killbill/billing/overdue/glue/TestOverdueModule.java
@@ -27,8 +27,10 @@ import org.killbill.billing.overdue.TestOverdueHelper;
 import org.killbill.billing.overdue.applicator.OverdueBusListenerTester;
 import org.killbill.billing.overdue.caching.EhCacheOverdueConfigCache;
 import org.killbill.billing.overdue.caching.MockOverdueConfigCache;
+import org.killbill.billing.overdue.caching.OverdueCacheInvalidationCallback;
 import org.killbill.billing.overdue.caching.OverdueConfigCache;
 import org.killbill.billing.platform.api.KillbillConfigSource;
+import org.killbill.billing.tenant.api.TenantInternalApi.CacheInvalidationCallback;
 import org.killbill.billing.util.email.EmailModule;
 import org.killbill.billing.util.email.templates.TemplateModule;
 import org.killbill.billing.util.glue.AuditModule;
@@ -36,6 +38,8 @@ import org.killbill.billing.util.glue.CacheModule;
 import org.killbill.billing.util.glue.CallContextModule;
 import org.killbill.billing.util.glue.CustomFieldModule;
 
+import com.google.inject.name.Names;
+
 public class TestOverdueModule extends DefaultOverdueModule {
 
     public TestOverdueModule(final KillbillConfigSource configSource) {
@@ -68,6 +72,7 @@ public class TestOverdueModule extends DefaultOverdueModule {
 
     public void installOverdueConfigCache() {
         bind(OverdueConfigCache.class).to(MockOverdueConfigCache.class).asEagerSingleton();
+        bind(CacheInvalidationCallback.class).annotatedWith(Names.named(OVERDUE_INVALIDATION_CALLBACK)).to(OverdueCacheInvalidationCallback.class).asEagerSingleton();
     }
 
 }

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 579b153..ee33163 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
@@ -24,7 +24,6 @@ import javax.inject.Inject;
 import javax.inject.Named;
 
 import org.killbill.billing.callcontext.InternalTenantContext;
-import org.killbill.billing.tenant.api.TenantCacheInvalidation.CacheInvalidationKey;
 import org.killbill.billing.tenant.api.TenantKV.TenantKey;
 import org.killbill.billing.tenant.dao.TenantDao;
 import org.killbill.billing.tenant.glue.DefaultTenantModule;
@@ -32,14 +31,14 @@ import org.killbill.billing.util.LocaleUtils;
 
 /**
  * This is the private API which is used to extract per tenant objects (catalog, overdue, invoice templates, ..)
- * <p>
+ * <p/>
  * Some of these per tenant objects are cached at a higher level in their respective modules (catalog, overdue) to
  * avoid reconstructing the object state from the xml definition each time. As a result, the module also registers
  * a callback which is used for the cache invalidation when the state changes and the operation occurred on a remote node.
  * For those objects, the private api is called from the module.
- * <p>
+ * <p/>
  * Some others (invoice templates,...) are not cached (yet) and so the logic is simpler.
- * <p>
+ * <p/>
  * The api can only be used to retrieve objects where no caching is required.
  */
 public class DefaultTenantInternalApi implements TenantInternalApi {
@@ -47,7 +46,6 @@ public class DefaultTenantInternalApi implements TenantInternalApi {
     private final TenantDao tenantDao;
     private final TenantCacheInvalidation tenantCacheInvalidation;
 
-
     @Inject
     public DefaultTenantInternalApi(@Named(DefaultTenantModule.NO_CACHING_TENANT) final TenantDao tenantDao,
                                     final TenantCacheInvalidation tenantCacheInvalidation) {
@@ -56,14 +54,17 @@ public class DefaultTenantInternalApi implements TenantInternalApi {
     }
 
     @Override
-    public List<String> getTenantCatalogs(final InternalTenantContext tenantContext, final CacheInvalidationCallback cacheInvalidationCallback) {
-        tenantCacheInvalidation.registerCallback(new CacheInvalidationKey(tenantContext.getTenantRecordId(), TenantKey.CATALOG), cacheInvalidationCallback);
+    public void initializeCacheInvalidationCallback(final TenantKey key, final CacheInvalidationCallback cacheInvalidationCallback) {
+        tenantCacheInvalidation.registerCallback(key, cacheInvalidationCallback);
+    }
+
+    @Override
+    public List<String> getTenantCatalogs(final InternalTenantContext tenantContext) {
         return tenantDao.getTenantValueForKey(TenantKey.CATALOG.toString(), tenantContext);
     }
 
     @Override
-    public String getTenantOverdueConfig(final InternalTenantContext tenantContext, final CacheInvalidationCallback cacheInvalidationCallback) {
-        tenantCacheInvalidation.registerCallback(new CacheInvalidationKey(tenantContext.getTenantRecordId(), TenantKey.OVERDUE_CONFIG), cacheInvalidationCallback);
+    public String getTenantOverdueConfig(final InternalTenantContext tenantContext) {
         final List<String> values = tenantDao.getTenantValueForKey(TenantKey.OVERDUE_CONFIG.toString(), tenantContext);
         return getUniqueValue(values, "overdue config", tenantContext);
     }

diff --git a/tenant/src/main/java/org/killbill/billing/tenant/api/TenantCacheInvalidation.java b/tenant/src/main/java/org/killbill/billing/tenant/api/TenantCacheInvalidation.java
index ec4ef07..d686fa8 100644
--- a/tenant/src/main/java/org/killbill/billing/tenant/api/TenantCacheInvalidation.java
+++ b/tenant/src/main/java/org/killbill/billing/tenant/api/TenantCacheInvalidation.java
@@ -50,9 +50,10 @@ import org.slf4j.LoggerFactory;
 public class TenantCacheInvalidation {
 
     private final static int TERMINATION_TIMEOUT_SEC = 5;
+
     private static final Logger logger = LoggerFactory.getLogger(TenantCacheInvalidation.class);
 
-    private final Map<CacheInvalidationKey, CacheInvalidationCallback> cache;
+    private final Map<TenantKey, CacheInvalidationCallback> cache;
     private final TenantBroadcastDao broadcastDao;
     private final ScheduledExecutorService tenantExecutor;
     private final TenantConfig tenantConfig;
@@ -63,7 +64,7 @@ public class TenantCacheInvalidation {
     public TenantCacheInvalidation(@Named(DefaultTenantModule.NO_CACHING_TENANT) final TenantBroadcastDao broadcastDao,
                                    @Named(DefaultTenantModule.TENANT_EXECUTOR_NAMED) final ScheduledExecutorService tenantExecutor,
                                    final TenantConfig tenantConfig) {
-        this.cache = new HashMap<CacheInvalidationKey, CacheInvalidationCallback>();
+        this.cache = new HashMap<TenantKey, CacheInvalidationCallback>();
         this.broadcastDao = broadcastDao;
         this.tenantExecutor = tenantExecutor;
         this.tenantConfig = tenantConfig;
@@ -106,13 +107,13 @@ public class TenantCacheInvalidation {
         }
     }
 
-    public void registerCallback(final CacheInvalidationKey key, final CacheInvalidationCallback value) {
+    public void registerCallback(final TenantKey key, final CacheInvalidationCallback value) {
         if (!cache.containsKey(key)) {
             cache.put(key, value);
         }
     }
 
-    public CacheInvalidationCallback getCacheInvalidation(final CacheInvalidationKey key) {
+    public CacheInvalidationCallback getCacheInvalidation(final TenantKey key) {
         return cache.get(key);
     }
 
@@ -150,59 +151,15 @@ public class TenantCacheInvalidation {
                     return;
                 }
 
-                final CacheInvalidationKey key = new CacheInvalidationKey(cur.getTenantRecordId(), TenantKey.valueOf(cur.getType()));
-                final CacheInvalidationCallback callback = parent.getCacheInvalidation(key);
+                final CacheInvalidationCallback callback = parent.getCacheInvalidation(TenantKey.valueOf(cur.getType()));
                 if (callback != null) {
                     final InternalTenantContext tenantContext = new InternalTenantContext(cur.getTenantRecordId(), null);
                     callback.invalidateCache(tenantContext);
+                } else {
+                    logger.warn("Failed to find CacheInvalidationCallback for " + cur.getType());
                 }
                 parent.setLatestRecordIdProcessed(cur.getRecordId());
             }
         }
     }
-
-    public static final class CacheInvalidationKey {
-
-        private final Long tenantRecordId;
-        private final TenantKey type;
-
-        public CacheInvalidationKey(final Long tenantRecordId, final TenantKey type) {
-            this.tenantRecordId = tenantRecordId;
-            this.type = type;
-        }
-
-        public Long getTenantRecordId() {
-            return tenantRecordId;
-        }
-
-        public TenantKey getType() {
-            return type;
-        }
-
-        @Override
-        public boolean equals(final Object o) {
-            if (this == o) {
-                return true;
-            }
-            if (!(o instanceof CacheInvalidationKey)) {
-                return false;
-            }
-
-            final CacheInvalidationKey that = (CacheInvalidationKey) o;
-            if (tenantRecordId != null ? !tenantRecordId.equals(that.tenantRecordId) : that.tenantRecordId != null) {
-                return false;
-            }
-            if (type != that.type) {
-                return false;
-            }
-            return true;
-        }
-
-        @Override
-        public int hashCode() {
-            int result = tenantRecordId != null ? tenantRecordId.hashCode() : 0;
-            result = 31 * result + (type != null ? type.hashCode() : 0);
-            return result;
-        }
-    }
 }

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
index cd82a27..8630e4d 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/TenantCatalogCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/TenantCatalogCacheLoader.java
@@ -27,10 +27,14 @@ import org.killbill.billing.catalog.api.CatalogApiException;
 import org.killbill.billing.tenant.api.TenantInternalApi;
 import org.killbill.billing.tenant.api.TenantInternalApi.CacheInvalidationCallback;
 import org.killbill.billing.util.cache.Cachable.CacheType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Singleton
 public class TenantCatalogCacheLoader extends BaseCacheLoader {
 
+    private final Logger logger = LoggerFactory.getLogger(TenantCatalogCacheLoader.class);
+
     private final TenantInternalApi tenantApi;
 
     @Inject
@@ -63,11 +67,12 @@ public class TenantCatalogCacheLoader extends BaseCacheLoader {
         }
 
         final LoaderCallback callback = (LoaderCallback) cacheLoaderArgument.getArgs()[0];
-        final List<String> catalogXMLs = tenantApi.getTenantCatalogs(internalTenantContext, (CacheInvalidationCallback) callback);
+        final List<String> catalogXMLs = tenantApi.getTenantCatalogs(internalTenantContext);
         if (catalogXMLs.isEmpty()) {
             return null;
         }
         try {
+            logger.info("Loading catalog cache for tenant " + internalTenantContext.getTenantRecordId());
             return callback.loadCatalog(catalogXMLs);
         } catch (CatalogApiException e) {
             throw new IllegalStateException(String.format("Failed to de-serialize catalog for tenant %s : %s",
@@ -75,7 +80,7 @@ public class TenantCatalogCacheLoader extends BaseCacheLoader {
         }
     }
 
-    public interface LoaderCallback extends CacheInvalidationCallback {
+    public interface LoaderCallback {
         public Object loadCatalog(final List<String> catalogXMLs) throws CatalogApiException;
     }
 }

diff --git a/util/src/main/java/org/killbill/billing/util/cache/TenantOverdueConfigCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/TenantOverdueConfigCacheLoader.java
index edcf7a4..7b53bb8 100644
--- a/util/src/main/java/org/killbill/billing/util/cache/TenantOverdueConfigCacheLoader.java
+++ b/util/src/main/java/org/killbill/billing/util/cache/TenantOverdueConfigCacheLoader.java
@@ -25,10 +25,14 @@ import org.killbill.billing.overdue.api.OverdueApiException;
 import org.killbill.billing.tenant.api.TenantInternalApi;
 import org.killbill.billing.tenant.api.TenantInternalApi.CacheInvalidationCallback;
 import org.killbill.billing.util.cache.Cachable.CacheType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Singleton
 public class TenantOverdueConfigCacheLoader extends BaseCacheLoader {
 
+    private static final Logger logger = LoggerFactory.getLogger(TenantOverdueConfigCacheLoader.class);
+
     private final TenantInternalApi tenantApi;
 
     @Inject
@@ -61,11 +65,13 @@ public class TenantOverdueConfigCacheLoader extends BaseCacheLoader {
         }
 
         final LoaderCallback callback = (LoaderCallback) cacheLoaderArgument.getArgs()[0];
-        final String overdueXML = tenantApi.getTenantOverdueConfig(internalTenantContext, (CacheInvalidationCallback) callback);
+        final String overdueXML = tenantApi.getTenantOverdueConfig(internalTenantContext);
         if (overdueXML == null) {
             return null;
         }
         try {
+            logger.info("Loading overdue cache for tenant " + internalTenantContext.getTenantRecordId());
+
             return callback.loadCatalog(overdueXML);
         } catch (OverdueApiException e) {
             throw new IllegalStateException(String.format("Failed to de-serialize overdue config for tenant %s : %s",
@@ -73,7 +79,7 @@ public class TenantOverdueConfigCacheLoader extends BaseCacheLoader {
         }
     }
 
-    public interface LoaderCallback extends CacheInvalidationCallback {
+    public interface LoaderCallback {
         public Object loadCatalog(final String overdueXML) throws OverdueApiException;
     }
 }

diff --git a/util/src/main/java/org/killbill/billing/util/config/TenantConfig.java b/util/src/main/java/org/killbill/billing/util/config/TenantConfig.java
index 48374c0..5eec621 100644
--- a/util/src/main/java/org/killbill/billing/util/config/TenantConfig.java
+++ b/util/src/main/java/org/killbill/billing/util/config/TenantConfig.java
@@ -26,7 +26,7 @@ public interface TenantConfig extends KillbillConfig {
 
     @Config("org.killbill.tenant.broadcast.rate")
     @Default("5s")
-    @Description("Rate at which janitor tasks are scheduled")
+    @Description("Rate at which tenant broadcast task is scheduled")
     public TimeSpan getTenantBroadcastServiceRunningRate();
 
 }