killbill-aplcache

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 6480aea..785d369 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,6 +22,7 @@ import java.util.Locale;
 
 import org.killbill.billing.callcontext.InternalTenantContext;
 import org.killbill.billing.tenant.api.TenantKV.TenantKey;
+import org.killbill.billing.util.callcontext.TenantContext;
 
 public interface TenantInternalApi {
 
@@ -42,4 +43,8 @@ public interface TenantInternalApi {
     public String getInvoiceTranslation(Locale locale, InternalTenantContext tenantContext);
 
     public String getCatalogTranslation(Locale locale, InternalTenantContext tenantContext);
+
+    public String getPluginConfig(String pluginName, InternalTenantContext tenantContext);
+
+    public List<String> getTenantValueForKey(final String key, final InternalTenantContext tenantContext);
 }

diff --git a/jaxrs/src/main/java/org/killbill/billing/jaxrs/resources/JaxrsResource.java b/jaxrs/src/main/java/org/killbill/billing/jaxrs/resources/JaxrsResource.java
index 0ff5408..5312673 100644
--- a/jaxrs/src/main/java/org/killbill/billing/jaxrs/resources/JaxrsResource.java
+++ b/jaxrs/src/main/java/org/killbill/billing/jaxrs/resources/JaxrsResource.java
@@ -28,6 +28,7 @@ public interface JaxrsResource {
 
     public static final String TIMELINE = "timeline";
     public static final String REGISTER_NOTIFICATION_CALLBACK = "registerNotificationCallback";
+    public static final String UPLOAD_PLUGIN_CONFIG = "uploadPluginConfig";
     public static final String SEARCH = "search";
 
     /*

diff --git a/jaxrs/src/main/java/org/killbill/billing/jaxrs/resources/TenantResource.java b/jaxrs/src/main/java/org/killbill/billing/jaxrs/resources/TenantResource.java
index 045eb04..23b8636 100644
--- a/jaxrs/src/main/java/org/killbill/billing/jaxrs/resources/TenantResource.java
+++ b/jaxrs/src/main/java/org/killbill/billing/jaxrs/resources/TenantResource.java
@@ -20,6 +20,7 @@ import java.net.URI;
 import java.util.List;
 import java.util.UUID;
 
+import javax.annotation.Nullable;
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
@@ -37,12 +38,11 @@ import javax.ws.rs.core.UriInfo;
 
 import org.killbill.billing.ObjectType;
 import org.killbill.billing.account.api.AccountUserApi;
-import org.killbill.billing.payment.api.PaymentApi;
-import org.killbill.clock.Clock;
 import org.killbill.billing.jaxrs.json.TenantJson;
 import org.killbill.billing.jaxrs.json.TenantKeyJson;
 import org.killbill.billing.jaxrs.util.Context;
 import org.killbill.billing.jaxrs.util.JaxrsUriBuilder;
+import org.killbill.billing.payment.api.PaymentApi;
 import org.killbill.billing.tenant.api.Tenant;
 import org.killbill.billing.tenant.api.TenantApiException;
 import org.killbill.billing.tenant.api.TenantData;
@@ -53,6 +53,7 @@ import org.killbill.billing.util.api.CustomFieldUserApi;
 import org.killbill.billing.util.api.TagUserApi;
 import org.killbill.billing.util.callcontext.CallContext;
 import org.killbill.billing.util.callcontext.TenantContext;
+import org.killbill.clock.Clock;
 
 import com.codahale.metrics.annotation.Timed;
 import com.google.inject.Inject;
@@ -135,16 +136,12 @@ public class TenantResource extends JaxRsResourceBase {
     @Produces(APPLICATION_JSON)
     @ApiOperation(value = "Create a push notification")
     @ApiResponses(value = {@ApiResponse(code = 400, message = "Invalid tenantId supplied")})
-    public Response registerPushNotificationCallback(@PathParam("tenantId") final String tenantId,
-                                                     @QueryParam(QUERY_NOTIFICATION_CALLBACK) final String notificationCallback,
+    public Response registerPushNotificationCallback(@QueryParam(QUERY_NOTIFICATION_CALLBACK) final String notificationCallback,
                                                      @HeaderParam(HDR_CREATED_BY) final String createdBy,
                                                      @HeaderParam(HDR_REASON) final String reason,
                                                      @HeaderParam(HDR_COMMENT) final String comment,
                                                      @javax.ws.rs.core.Context final HttpServletRequest request) throws TenantApiException {
-        final CallContext callContext = context.createContext(createdBy, reason, comment, request);
-        tenantApi.addTenantKeyValue(TenantKey.PUSH_NOTIFICATION_CB.toString(), notificationCallback, callContext);
-        final URI uri = UriBuilder.fromResource(TenantResource.class).path(TenantResource.class, "getPushNotificationCallbacks").build();
-        return Response.created(uri).build();
+        return insertTenantKey(TenantKey.PUSH_NOTIFICATION_CB, null, notificationCallback, "getPushNotificationCallbacks", createdBy, reason, comment, request);
     }
 
     @Timed
@@ -154,25 +151,96 @@ public class TenantResource extends JaxRsResourceBase {
     @ApiOperation(value = "Retrieve a push notification", response = TenantKeyJson.class)
     @ApiResponses(value = {@ApiResponse(code = 400, message = "Invalid tenantId supplied")})
     public Response getPushNotificationCallbacks(@javax.ws.rs.core.Context final HttpServletRequest request) throws TenantApiException {
-
-        final TenantContext tenatContext = context.createContext(request);
-        final List<String> values = tenantApi.getTenantValueForKey(TenantKey.PUSH_NOTIFICATION_CB.toString(), tenatContext);
-        final TenantKeyJson result = new TenantKeyJson(TenantKey.PUSH_NOTIFICATION_CB.toString(), values);
-        return Response.status(Status.OK).entity(result).build();
+        return getTenantKey(TenantKey.PUSH_NOTIFICATION_CB, null, request);
     }
 
     @Timed
     @DELETE
     @Path("/REGISTER_NOTIFICATION_CALLBACK")
+    //@Path("/" + REGISTER_NOTIFICATION_CALLBACK) @see #238
     @ApiOperation(value = "Delete a push notification")
     @ApiResponses(value = {@ApiResponse(code = 400, message = "Invalid tenantId supplied")})
-    public Response deletePushNotificationCallbacks(@PathParam("tenantId") final String tenantId,
-                                                    @HeaderParam(HDR_CREATED_BY) final String createdBy,
+    public Response deletePushNotificationCallbacks(@HeaderParam(HDR_CREATED_BY) final String createdBy,
                                                     @HeaderParam(HDR_REASON) final String reason,
                                                     @HeaderParam(HDR_COMMENT) final String comment,
                                                     @javax.ws.rs.core.Context final HttpServletRequest request) throws TenantApiException {
+        return deleteTenantKey(TenantKey.PUSH_NOTIFICATION_CB, null, createdBy, reason, comment, request);
+    }
+
+    @Timed
+    @POST
+    @Path("/" + UPLOAD_PLUGIN_CONFIG + "/{pluginName:" + ANYTHING_PATTERN + "}")
+    @Consumes(APPLICATION_JSON)
+    @Produces(APPLICATION_JSON)
+    @ApiOperation(value = "Add a per tenant configuration for a plugin")
+    @ApiResponses(value = {@ApiResponse(code = 400, message = "Invalid tenantId supplied")})
+    public Response uploadPluginConfiguration(final String pluginConfig,
+                                              @PathParam("pluginName") final String pluginName,
+                                              @HeaderParam(HDR_CREATED_BY) final String createdBy,
+                                              @HeaderParam(HDR_REASON) final String reason,
+                                              @HeaderParam(HDR_COMMENT) final String comment,
+                                              @javax.ws.rs.core.Context final HttpServletRequest request) throws TenantApiException {
+        return insertTenantKey(TenantKey.PLUGIN_CONFIG_, pluginName, pluginConfig, "getPluginConfiguration", createdBy, reason, comment, request);
+    }
+
+    @Timed
+    @GET
+    @Path("/" + UPLOAD_PLUGIN_CONFIG + "/{pluginName:" + ANYTHING_PATTERN + "}")
+    @Produces(APPLICATION_JSON)
+    @ApiOperation(value = "Retrieve a per tenant configuration for a plugin", response = TenantKeyJson.class)
+    @ApiResponses(value = {@ApiResponse(code = 400, message = "Invalid tenantId supplied")})
+    public Response getPluginConfiguration(@PathParam("pluginName") final String pluginName,
+                                           @javax.ws.rs.core.Context final HttpServletRequest request) throws TenantApiException {
+        return getTenantKey(TenantKey.PLUGIN_CONFIG_, pluginName, request);
+    }
+
+    @Timed
+    @DELETE
+    @Path("/" + UPLOAD_PLUGIN_CONFIG + "/{pluginName:" + ANYTHING_PATTERN + "}")
+    @ApiOperation(value = "Delete a per tenant configuration for a plugin")
+    @ApiResponses(value = {@ApiResponse(code = 400, message = "Invalid tenantId supplied")})
+    public Response deletePluginConfiguration(@PathParam("pluginName") final String pluginName,
+                                              @HeaderParam(HDR_CREATED_BY) final String createdBy,
+                                              @HeaderParam(HDR_REASON) final String reason,
+                                              @HeaderParam(HDR_COMMENT) final String comment,
+                                              @javax.ws.rs.core.Context final HttpServletRequest request) throws TenantApiException {
+        return deleteTenantKey(TenantKey.PLUGIN_CONFIG_, pluginName, createdBy, reason, comment, request);
+    }
+
+    private Response insertTenantKey(final TenantKey key,
+                                     @Nullable final String keyPostfix,
+                                     final String value,
+                                     final String getMethodStr,
+                                     final String createdBy,
+                                     final String reason,
+                                     final String comment,
+                                     final HttpServletRequest request) throws TenantApiException {
+        final CallContext callContext = context.createContext(createdBy, reason, comment, request);
+        final String tenantKey = keyPostfix != null ? key.toString() + keyPostfix : key.toString();
+        tenantApi.addTenantKeyValue(tenantKey, value, callContext);
+        final URI uri = UriBuilder.fromResource(TenantResource.class).path(TenantResource.class, getMethodStr).build();
+        return Response.created(uri).build();
+    }
+
+    private Response getTenantKey(final TenantKey key,
+                                  @Nullable final String keyPostfix,
+                                  final HttpServletRequest request) throws TenantApiException {
+        final TenantContext tenantContext = context.createContext(request);
+        final String tenantKey = keyPostfix != null ? key.toString() + keyPostfix : key.toString();
+        final List<String> values = tenantApi.getTenantValueForKey(tenantKey, tenantContext);
+        final TenantKeyJson result = new TenantKeyJson(tenantKey, values);
+        return Response.status(Status.OK).entity(result).build();
+    }
+
+    private Response deleteTenantKey(final TenantKey key,
+                                     @Nullable final String keyPostfix,
+                                     final String createdBy,
+                                     final String reason,
+                                     final String comment,
+                                     final HttpServletRequest request) throws TenantApiException {
         final CallContext callContext = context.createContext(createdBy, reason, comment, request);
-        tenantApi.deleteTenantKey(TenantKey.PUSH_NOTIFICATION_CB.toString(), callContext);
+        final String tenantKey = keyPostfix != null ? key.toString() + keyPostfix : key.toString();
+        tenantApi.deleteTenantKey(tenantKey, callContext);
         return Response.status(Status.OK).build();
     }
 

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 ee33163..ebfa5fd 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
@@ -93,6 +93,20 @@ public class DefaultTenantInternalApi implements TenantInternalApi {
         return getUniqueValue(values, "catalog translation", tenantContext);
     }
 
+    @Override
+    public String getPluginConfig(final String pluginName, final InternalTenantContext tenantContext) {
+        final String pluginConfigKey = TenantKey.PLUGIN_CONFIG_ + pluginName;
+        final List<String> values = tenantDao.getTenantValueForKey(pluginConfigKey, tenantContext);
+        return getUniqueValue(values, "config for plugin " + pluginConfigKey, tenantContext);
+    }
+
+    @Override
+    public List<String> getTenantValueForKey(final String key, final InternalTenantContext tenantContext) {
+        return tenantDao.getTenantValueForKey(key, tenantContext);
+    }
+
+
+
     private String getUniqueValue(final List<String> values, final String msg, final InternalTenantContext tenantContext) {
         if (values.isEmpty()) {
             return null;

diff --git a/tenant/src/main/java/org/killbill/billing/tenant/api/user/DefaultTenantUserApi.java b/tenant/src/main/java/org/killbill/billing/tenant/api/user/DefaultTenantUserApi.java
index f5781f7..8e72497 100644
--- a/tenant/src/main/java/org/killbill/billing/tenant/api/user/DefaultTenantUserApi.java
+++ b/tenant/src/main/java/org/killbill/billing/tenant/api/user/DefaultTenantUserApi.java
@@ -20,30 +20,42 @@ import java.util.List;
 import java.util.UUID;
 
 import org.killbill.billing.ErrorCode;
+import org.killbill.billing.ObjectType;
+import org.killbill.billing.callcontext.InternalCallContext;
+import org.killbill.billing.callcontext.InternalTenantContext;
 import org.killbill.billing.tenant.api.DefaultTenant;
 import org.killbill.billing.tenant.api.Tenant;
 import org.killbill.billing.tenant.api.TenantApiException;
 import org.killbill.billing.tenant.api.TenantData;
+import org.killbill.billing.tenant.api.TenantKV.TenantKey;
 import org.killbill.billing.tenant.api.TenantUserApi;
 import org.killbill.billing.tenant.dao.TenantDao;
 import org.killbill.billing.tenant.dao.TenantModelDao;
+import org.killbill.billing.util.cache.Cachable.CacheType;
+import org.killbill.billing.util.cache.CacheController;
+import org.killbill.billing.util.cache.CacheControllerDispatcher;
+import org.killbill.billing.util.cache.CacheLoaderArgument;
 import org.killbill.billing.util.callcontext.CallContext;
-import org.killbill.billing.callcontext.InternalCallContext;
 import org.killbill.billing.util.callcontext.InternalCallContextFactory;
-import org.killbill.billing.callcontext.InternalTenantContext;
 import org.killbill.billing.util.callcontext.TenantContext;
 
+import com.google.common.base.Predicate;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
 import com.google.inject.Inject;
 
 public class DefaultTenantUserApi implements TenantUserApi {
 
     private final TenantDao tenantDao;
     private final InternalCallContextFactory internalCallContextFactory;
+    private final CacheController<Object, Object> tenantKVCache;
 
     @Inject
-    public DefaultTenantUserApi(final TenantDao tenantDao, final InternalCallContextFactory internalCallContextFactory) {
+    public DefaultTenantUserApi(final TenantDao tenantDao, final InternalCallContextFactory internalCallContextFactory, final CacheControllerDispatcher cacheControllerDispatcher) {
         this.tenantDao = tenantDao;
         this.internalCallContextFactory = internalCallContextFactory;
+        this.tenantKVCache = cacheControllerDispatcher.getCacheController(CacheType.TENANT_KV);
+
     }
 
     @Override
@@ -81,23 +93,30 @@ public class DefaultTenantUserApi implements TenantUserApi {
     @Override
     public List<String> getTenantValueForKey(final String key, final TenantContext context)
             throws TenantApiException {
+
         final InternalTenantContext internalContext = internalCallContextFactory.createInternalTenantContext(context);
+        final String value = getCachedTenantValueForKey(key, internalContext);
+        if (value != null) {
+            return ImmutableList.<String>of(value);
+        }
         return tenantDao.getTenantValueForKey(key, internalContext);
     }
 
     @Override
     public void addTenantKeyValue(final String key, final String value, final CallContext context)
             throws TenantApiException {
-
-        final InternalCallContext internalContext = internalCallContextFactory.createInternalCallContext(context);
-        // TODO Figure out the exact verification if nay
         /*
-        final Tenant tenant = tenantDao.getById(callcontext.getTenantId(), internalContext);
+        final Tenant tenant = tenantDao.getById(tenantId, new InternalTenantContext(null, null));
         if (tenant == null) {
             throw new TenantApiException(ErrorCode.TENANT_DOES_NOT_EXIST_FOR_ID, tenantId);
         }
         */
-        tenantDao.addTenantKeyValue(key, value, internalContext);
+
+        final InternalCallContext internalContext = internalCallContextFactory.createInternalCallContext(context);
+        final String tenantKey = getCacheKeyName(key, internalContext);
+        // Invalidate tenantKVCache before we store. Multi-node invalidation will follow the TenantBroadcast pattern
+        tenantKVCache.remove(tenantKey);
+        tenantDao.addTenantKeyValue(key, value, isSingleValueKey(key), internalContext);
     }
 
     @Override
@@ -109,7 +128,36 @@ public class DefaultTenantUserApi implements TenantUserApi {
             throw new TenantApiException(ErrorCode.TENANT_DOES_NOT_EXIST_FOR_ID, tenantId);
         }
         */
+
+        // Invalidate tenantKVCache before we store. Multi-node invalidation will follow the TenantBroadcast pattern
         final InternalCallContext internalContext = internalCallContextFactory.createInternalCallContext(context);
+        final String tenantKey = getCacheKeyName(key, internalContext);
+        tenantKVCache.remove(tenantKey);
         tenantDao.deleteTenantKey(key, internalContext);
     }
+
+    private String getCachedTenantValueForKey(final String key, final InternalTenantContext internalContext) {
+
+        if (!isSingleValueKey(key)) {
+            return null;
+        }
+        final String tenantKey = getCacheKeyName(key, internalContext);
+        return (String) tenantKVCache.get(tenantKey, new CacheLoaderArgument(ObjectType.TENANT_KVS));
+    }
+
+    private String getCacheKeyName(final String key, final InternalTenantContext internalContext) {
+        final StringBuilder tenantKey = new StringBuilder(key);
+        tenantKey.append(CacheControllerDispatcher.CACHE_KEY_SEPARATOR);
+        tenantKey.append(internalContext.getTenantRecordId());
+        return tenantKey.toString();
+    }
+
+    private boolean isSingleValueKey(final String key) {
+        return Iterables.tryFind(ImmutableList.copyOf(TenantKey.values()), new Predicate<TenantKey>() {
+            @Override
+            public boolean apply(final TenantKey input) {
+                return input.isSingleValue() && key.startsWith(input.toString());
+            }
+        }).orNull() != null;
+    }
 }

diff --git a/tenant/src/main/java/org/killbill/billing/tenant/dao/DefaultTenantDao.java b/tenant/src/main/java/org/killbill/billing/tenant/dao/DefaultTenantDao.java
index 77c6df6..2b5b55a 100644
--- a/tenant/src/main/java/org/killbill/billing/tenant/dao/DefaultTenantDao.java
+++ b/tenant/src/main/java/org/killbill/billing/tenant/dao/DefaultTenantDao.java
@@ -46,8 +46,10 @@ import org.killbill.billing.util.entity.dao.EntitySqlDaoWrapperFactory;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Function;
+import com.google.common.base.Predicate;
 import com.google.common.collect.Collections2;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
 import com.google.inject.Inject;
 
 public class DefaultTenantDao extends EntityDaoBase<TenantModelDao, Tenant, TenantApiException> implements TenantDao {
@@ -126,12 +128,16 @@ public class DefaultTenantDao extends EntityDaoBase<TenantModelDao, Tenant, Tena
     }
 
     @Override
-    public void addTenantKeyValue(final String key, final String value, final InternalCallContext context) {
+    public void addTenantKeyValue(final String key, final String value, final boolean uniqueKey, final InternalCallContext context) {
         transactionalSqlDao.execute(new EntitySqlDaoTransactionWrapper<Void>() {
             @Override
             public Void inTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory) throws Exception {
                 final TenantKVModelDao tenantKVModelDao = new TenantKVModelDao(UUID.randomUUID(), context.getCreatedDate(), context.getUpdatedDate(), key, value);
-                entitySqlDaoWrapperFactory.become(TenantKVSqlDao.class).create(tenantKVModelDao, context);
+                final TenantKVSqlDao tenantKVSqlDao = entitySqlDaoWrapperFactory.become(TenantKVSqlDao.class);
+                if (uniqueKey) {
+                    deleteFromTransaction(key, entitySqlDaoWrapperFactory, context);
+                }
+                tenantKVSqlDao.create(tenantKVModelDao, context);
                 broadcastConfigurationChangeFromTransaction(entitySqlDaoWrapperFactory, key, context);
                 return null;
             }
@@ -144,23 +150,41 @@ public class DefaultTenantDao extends EntityDaoBase<TenantModelDao, Tenant, Tena
         transactionalSqlDao.execute(new EntitySqlDaoTransactionWrapper<Void>() {
             @Override
             public Void inTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory) throws Exception {
-                final List<TenantKVModelDao> tenantKVs = entitySqlDaoWrapperFactory.become(TenantKVSqlDao.class).getTenantValueForKey(key, context);
-                for (TenantKVModelDao cur : tenantKVs) {
-                    if (cur.getTenantKey().equals(key)) {
-                        entitySqlDaoWrapperFactory.become(TenantKVSqlDao.class).markTenantKeyAsDeleted(cur.getId().toString(), context);
-                    }
-                }
-                return null;
+                broadcastConfigurationChangeFromTransaction(entitySqlDaoWrapperFactory, key, context);
+                return deleteFromTransaction(key, entitySqlDaoWrapperFactory, context);
             }
+
         });
     }
 
+    private Void deleteFromTransaction(final String key, final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory, final InternalCallContext context) {
+        final List<TenantKVModelDao> tenantKVs = entitySqlDaoWrapperFactory.become(TenantKVSqlDao.class).getTenantValueForKey(key, context);
+        for (TenantKVModelDao cur : tenantKVs) {
+            if (cur.getTenantKey().equals(key)) {
+                entitySqlDaoWrapperFactory.become(TenantKVSqlDao.class).markTenantKeyAsDeleted(cur.getId().toString(), context);
+            }
+        }
+        return null;
+    }
+
     private void broadcastConfigurationChangeFromTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory,
                                                              final String key, final InternalCallContext context) throws EntityPersistenceException {
-        if (key.equals(TenantKey.CATALOG.toString()) ||
-            key.equals(TenantKey.OVERDUE_CONFIG.toString())) {
+        if (isSystemKey(key)) {
             final TenantBroadcastModelDao broadcast = new TenantBroadcastModelDao(key);
             entitySqlDaoWrapperFactory.become(TenantBroadcastSqlDao.class).create(broadcast, context);
         }
     }
+
+    //
+    // For now we restrict the caching to the (system) TenantKey keys
+    //
+    private boolean isSystemKey(final String key) {
+        return Iterables.tryFind(ImmutableList.copyOf(TenantKey.values()), new Predicate<TenantKey>() {
+            @Override
+            public boolean apply(final TenantKey input) {
+                return key.startsWith(input.toString());
+            }
+        }).orNull() != null;
+    }
+
 }

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
index 7230082..711886d 100644
--- a/tenant/src/main/java/org/killbill/billing/tenant/dao/NoCachingTenantDao.java
+++ b/tenant/src/main/java/org/killbill/billing/tenant/dao/NoCachingTenantDao.java
@@ -17,6 +17,7 @@
 
 package org.killbill.billing.tenant.dao;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 
@@ -65,7 +66,7 @@ public class NoCachingTenantDao extends EntityDaoBase<TenantModelDao, Tenant, Te
             @Override
             public List<String> inTransaction(final EntitySqlDaoWrapperFactory entitySqlDaoWrapperFactory) throws Exception {
                 final List<TenantKVModelDao> tenantKV = entitySqlDaoWrapperFactory.become(TenantKVSqlDao.class).getTenantValueForKey(key, context);
-                return ImmutableList.copyOf(Collections2.transform(tenantKV, new Function<TenantKVModelDao, String>() {
+                return new ArrayList(Collections2.transform(tenantKV, new Function<TenantKVModelDao, String>() {
                     @Override
                     public String apply(final TenantKVModelDao in) {
                         return in.getTenantValue();
@@ -76,7 +77,7 @@ public class NoCachingTenantDao extends EntityDaoBase<TenantModelDao, Tenant, Te
     }
 
     @Override
-    public void addTenantKeyValue(final String key, final String value, final InternalCallContext context) {
+    public void addTenantKeyValue(final String key, final String value, final boolean uniqueKey, final InternalCallContext context) {
         throw new IllegalStateException("Not implemented by NoCachingTenantDao");
     }
 

diff --git a/tenant/src/main/java/org/killbill/billing/tenant/dao/TenantDao.java b/tenant/src/main/java/org/killbill/billing/tenant/dao/TenantDao.java
index 84fbcde..7b32c57 100644
--- a/tenant/src/main/java/org/killbill/billing/tenant/dao/TenantDao.java
+++ b/tenant/src/main/java/org/killbill/billing/tenant/dao/TenantDao.java
@@ -30,7 +30,7 @@ public interface TenantDao extends EntityDao<TenantModelDao, Tenant, TenantApiEx
 
     public List<String> getTenantValueForKey(final String key, final InternalTenantContext context);
 
-    public void addTenantKeyValue(final String key, final String value, final InternalCallContext context);
+    public void addTenantKeyValue(final String key, final String value, final boolean uniqueKey, final InternalCallContext context);
 
     public void deleteTenantKey(final String key, final InternalCallContext context);
 }

diff --git a/tenant/src/test/java/org/killbill/billing/tenant/api/user/TestDefaultTenantUserApi.java b/tenant/src/test/java/org/killbill/billing/tenant/api/user/TestDefaultTenantUserApi.java
new file mode 100644
index 0000000..dfb73f7
--- /dev/null
+++ b/tenant/src/test/java/org/killbill/billing/tenant/api/user/TestDefaultTenantUserApi.java
@@ -0,0 +1,100 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.tenant.api.user;
+
+import java.util.List;
+
+import org.killbill.billing.tenant.TenantTestSuiteWithEmbeddedDb;
+import org.killbill.billing.tenant.api.TenantKV.TenantKey;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class TestDefaultTenantUserApi extends TenantTestSuiteWithEmbeddedDb {
+
+    @Test(groups = "slow")
+    public void testUserKey() throws Exception {
+        tenantUserApi.addTenantKeyValue("THE_KEY", "TheValue", callContext);
+
+        List<String> value = tenantUserApi.getTenantValueForKey("THE_KEY", callContext);
+        Assert.assertEquals(value.size(), 1);
+        Assert.assertEquals(value.get(0), "TheValue");
+
+        tenantUserApi.addTenantKeyValue("THE_KEY", "TheSecondValue", callContext);
+        value = tenantUserApi.getTenantValueForKey("THE_KEY", callContext);
+        Assert.assertEquals(value.size(), 2);
+
+        value = tenantUserApi.getTenantValueForKey("THE_KEY", callContext);
+        Assert.assertEquals(value.size(), 2);
+
+        tenantUserApi.deleteTenantKey("THE_KEY", callContext);
+        value = tenantUserApi.getTenantValueForKey("THE_KEY", callContext);
+        Assert.assertEquals(value.size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSystemKeySingleValue() throws Exception {
+
+        final String tenantKey = TenantKey.PLUGIN_CONFIG_.toString() + "MyPluginName";
+
+        tenantUserApi.addTenantKeyValue(tenantKey, "TheValue", callContext);
+
+        List<String> value = tenantUserApi.getTenantValueForKey(tenantKey, callContext);
+        Assert.assertEquals(value.size(), 1);
+        Assert.assertEquals(value.get(0), "TheValue");
+
+        // Warm cache
+        value = tenantUserApi.getTenantValueForKey(tenantKey, callContext);
+        Assert.assertEquals(value.size(), 1);
+        Assert.assertEquals(value.get(0), "TheValue");
+
+        tenantUserApi.addTenantKeyValue(tenantKey, "TheSecondValue", callContext);
+        value = tenantUserApi.getTenantValueForKey(tenantKey, callContext);
+        Assert.assertEquals(value.size(), 1);
+        Assert.assertEquals(value.get(0), "TheSecondValue");
+
+        // Warm cache
+        value = tenantUserApi.getTenantValueForKey(tenantKey, callContext);
+        Assert.assertEquals(value.size(), 1);
+        Assert.assertEquals(value.get(0), "TheSecondValue");
+
+        tenantUserApi.deleteTenantKey(tenantKey, callContext);
+        value = tenantUserApi.getTenantValueForKey(tenantKey, callContext);
+        Assert.assertEquals(value.size(), 0);
+    }
+
+    @Test(groups = "slow")
+    public void testSystemKeyMultipleValue() throws Exception {
+
+        final String tenantKey = TenantKey.CATALOG.toString();
+
+        tenantUserApi.addTenantKeyValue(tenantKey, "TheValue", callContext);
+
+        List<String> value = tenantUserApi.getTenantValueForKey(tenantKey, callContext);
+        Assert.assertEquals(value.size(), 1);
+        Assert.assertEquals(value.get(0), "TheValue");
+
+        tenantUserApi.addTenantKeyValue(tenantKey, "TheSecondValue", callContext);
+        value = tenantUserApi.getTenantValueForKey(tenantKey, callContext);
+        Assert.assertEquals(value.size(), 2);
+
+        tenantUserApi.deleteTenantKey(tenantKey, callContext);
+        value = tenantUserApi.getTenantValueForKey(tenantKey, callContext);
+        Assert.assertEquals(value.size(), 0);
+    }
+
+}

diff --git a/tenant/src/test/java/org/killbill/billing/tenant/dao/TestDefaultTenantDao.java b/tenant/src/test/java/org/killbill/billing/tenant/dao/TestDefaultTenantDao.java
index 48edb93..ddba113 100644
--- a/tenant/src/test/java/org/killbill/billing/tenant/dao/TestDefaultTenantDao.java
+++ b/tenant/src/test/java/org/killbill/billing/tenant/dao/TestDefaultTenantDao.java
@@ -59,18 +59,21 @@ public class TestDefaultTenantDao extends TenantTestSuiteWithEmbeddedDb {
                                                        UUID.randomUUID().toString(), UUID.randomUUID().toString());
         tenantDao.create(new TenantModelDao(tenant), internalCallContext);
 
-        tenantDao.addTenantKeyValue("TheKey", "TheValue", internalCallContext);
+        tenantDao .addTenantKeyValue("THE_KEY", "TheValue", false, internalCallContext);
 
-        List<String> value = tenantDao.getTenantValueForKey("TheKey", internalCallContext);
+        List<String> value = tenantDao.getTenantValueForKey("THE_KEY", internalCallContext);
         Assert.assertEquals(value.size(), 1);
         Assert.assertEquals(value.get(0), "TheValue");
 
-        tenantDao.addTenantKeyValue("TheKey", "TheSecondValue", internalCallContext);
-        value = tenantDao.getTenantValueForKey("TheKey", internalCallContext);
+        tenantDao.addTenantKeyValue("THE_KEY", "TheSecondValue", false, internalCallContext);
+        value = tenantDao.getTenantValueForKey("THE_KEY", internalCallContext);
         Assert.assertEquals(value.size(), 2);
 
-        tenantDao.deleteTenantKey("TheKey", internalCallContext);
-        value = tenantDao.getTenantValueForKey("TheKey", internalCallContext);
+        value = tenantDao.getTenantValueForKey("THE_KEY", internalCallContext);
+        Assert.assertEquals(value.size(), 2);
+
+        tenantDao.deleteTenantKey("THE_KEY", internalCallContext);
+        value = tenantDao.getTenantValueForKey("THE_KEY", internalCallContext);
         Assert.assertEquals(value.size(), 0);
     }
 }

diff --git a/tenant/src/test/java/org/killbill/billing/tenant/TenantTestSuiteWithEmbeddedDb.java b/tenant/src/test/java/org/killbill/billing/tenant/TenantTestSuiteWithEmbeddedDb.java
index 2a9429d..48b6287 100644
--- a/tenant/src/test/java/org/killbill/billing/tenant/TenantTestSuiteWithEmbeddedDb.java
+++ b/tenant/src/test/java/org/killbill/billing/tenant/TenantTestSuiteWithEmbeddedDb.java
@@ -18,6 +18,7 @@ package org.killbill.billing.tenant;
 
 import javax.inject.Named;
 
+import org.killbill.billing.tenant.api.TenantUserApi;
 import org.killbill.billing.tenant.dao.NoCachingTenantBroadcastDao;
 import org.killbill.billing.tenant.dao.TenantBroadcastDao;
 import org.killbill.billing.tenant.glue.DefaultTenantModule;
@@ -43,6 +44,9 @@ public class TenantTestSuiteWithEmbeddedDb extends GuicyKillbillTestSuiteWithEmb
     protected TenantBroadcastDao noCachingTenantBroadcastDao;
 
     @Inject
+    protected TenantUserApi tenantUserApi;
+
+    @Inject
     protected TenantBroadcastDao tenantBroadcastDao;
 
     @BeforeClass(groups = "slow")

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 024bdff..542b14e 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
@@ -33,6 +33,7 @@ public @interface Cachable {
     public final String AUDIT_LOG_VIA_HISTORY_CACHE_NAME = "audit-log-via-history";
     public final String TENANT_CATALOG_CACHE_NAME = "tenant-catalog";
     public final String TENANT_OVERDUE_CONFIG_CACHE_NAME = "tenant-overdue-config";
+    public final String TENANT_KV_CACHE_NAME = "tenant-kv";
 
     public CacheType value();
 
@@ -59,7 +60,10 @@ public @interface Cachable {
         TENANT_CATALOG(TENANT_CATALOG_CACHE_NAME, false),
 
         /* Tenant overdue config cache */
-        TENANT_OVERDUE_CONFIG(TENANT_OVERDUE_CONFIG_CACHE_NAME, false);
+        TENANT_OVERDUE_CONFIG(TENANT_OVERDUE_CONFIG_CACHE_NAME, false),
+
+        /* Tenant overdue config cache */
+        TENANT_KV(TENANT_KV_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 ab5eaa5..3698c4b 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
@@ -47,7 +47,8 @@ public class EhCacheCacheManagerProvider implements Provider<CacheManager> {
                                        final AuditLogCacheLoader auditLogCacheLoader,
                                        final AuditLogViaHistoryCacheLoader auditLogViaHistoryCacheLoader,
                                        final TenantCatalogCacheLoader tenantCatalogCacheLoader,
-                                       final TenantOverdueConfigCacheLoader tenantOverdueConfigCacheLoader) {
+                                       final TenantOverdueConfigCacheLoader tenantOverdueConfigCacheLoader,
+                                       final TenantKVCacheLoader tenantKVCacheLoader) {
         this.cacheConfig = cacheConfig;
         cacheLoaders.add(recordIdCacheLoader);
         cacheLoaders.add(accountRecordIdCacheLoader);
@@ -57,6 +58,7 @@ public class EhCacheCacheManagerProvider implements Provider<CacheManager> {
         cacheLoaders.add(auditLogViaHistoryCacheLoader);
         cacheLoaders.add(tenantCatalogCacheLoader);
         cacheLoaders.add(tenantOverdueConfigCacheLoader);
+        cacheLoaders.add(tenantKVCacheLoader);
     }
 
     @Override

diff --git a/util/src/main/java/org/killbill/billing/util/cache/TenantKVCacheLoader.java b/util/src/main/java/org/killbill/billing/util/cache/TenantKVCacheLoader.java
new file mode 100644
index 0000000..1e6cea2
--- /dev/null
+++ b/util/src/main/java/org/killbill/billing/util/cache/TenantKVCacheLoader.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2014-2015 Groupon, Inc
+ * Copyright 2014-2015 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.util.cache;
+
+import java.util.List;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import org.killbill.billing.callcontext.InternalTenantContext;
+import org.killbill.billing.tenant.api.TenantInternalApi;
+import org.killbill.billing.util.cache.Cachable.CacheType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Singleton
+public class TenantKVCacheLoader extends BaseCacheLoader {
+
+    private static final Logger logger = LoggerFactory.getLogger(TenantKVCacheLoader.class);
+    private final TenantInternalApi tenantApi;
+
+    @Inject
+    public TenantKVCacheLoader(final TenantInternalApi tenantApi) {
+        super();
+        this.tenantApi = tenantApi;
+    }
+
+    @Override
+    public CacheType getCacheType() {
+        return CacheType.TENANT_KV;
+    }
+
+    @Override
+    public Object load(final Object key, final Object argument) {
+
+        checkCacheLoaderStatus();
+
+        if (!(key instanceof String)) {
+            throw new IllegalArgumentException("Unexpected key type of " + key.getClass().getName());
+        }
+        if (!(argument instanceof CacheLoaderArgument)) {
+            throw new IllegalArgumentException("Unexpected key type of " + argument.getClass().getName());
+        }
+        final String[] parts = ((String) key).split(CacheControllerDispatcher.CACHE_KEY_SEPARATOR);
+        final String rawKey = parts[0];
+        final String tenantRecordId = parts[1];
+
+        final InternalTenantContext internalTenantContext = new InternalTenantContext(Long.valueOf(tenantRecordId));
+        final List<String> valuesForKey = tenantApi.getTenantValueForKey(rawKey, internalTenantContext);
+        if (valuesForKey == null || valuesForKey.size() == 0) {
+            return null;
+        }
+        if (valuesForKey.size() > 1) {
+            throw new IllegalStateException("TenantKVCacheLoader expecting no more than one value for key " + key);
+        }
+        return valuesForKey.get(0);
+    }
+}

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 44603dc..b35649c 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
@@ -189,7 +189,7 @@ public class EntitySqlDaoWrapperInvocationHandler<S extends EntitySqlDao<M, E>, 
         // This can't be AUDIT'ed and CACHABLE'd at the same time as we only cache 'get'
         if (auditedAnnotation != null) {
             return invokeWithAuditAndHistory(auditedAnnotation, method, args);
-        } else if (cachableAnnotation != null) {
+        } else if (cachableAnnotation != null && cacheControllerDispatcher != null) {
             return invokeWithCaching(cachableAnnotation, method, args);
         } else {
             return invokeRaw(method, args);

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