Merge pull request #40 from metamx/broker-cacheconfig

configuration option to switch cache implementations

client/src/main/java/com/metamx/druid/client/cache/CacheConfig.java

@@ -0,0 +1,11 @@
+package com.metamx.druid.client.cache;
+
+import org.skife.config.Config;
+import org.skife.config.Default;
+
+public abstract class CacheConfig
+{
+  @Config("druid.bard.cache.type")
+  @Default("local")
+  public abstract String getType();
+}

client/src/main/java/com/metamx/druid/http/BrokerNode.java

@@ -35,9 +35,12 @@ import com.metamx.druid.client.CachingClusteredClient;
 import com.metamx.druid.client.ClientConfig;
 import com.metamx.druid.client.ClientInventoryManager;
 import com.metamx.druid.client.cache.CacheBroker;
+import com.metamx.druid.client.cache.CacheConfig;
 import com.metamx.druid.client.cache.CacheMonitor;
 import com.metamx.druid.client.cache.MapCacheBroker;
 import com.metamx.druid.client.cache.MapCacheBrokerConfig;
+import com.metamx.druid.client.cache.MemcachedCacheBroker;
+import com.metamx.druid.client.cache.MemcachedCacheBrokerConfig;
 import com.metamx.druid.initialization.Initialization;
 import com.metamx.druid.initialization.ServiceDiscoveryConfig;
 import com.metamx.druid.jackson.DefaultObjectMapper;
@@ -66,6 +69,10 @@ public class BrokerNode extends QueryableNode<BrokerNode>
 {
   private static final Logger log = new Logger(BrokerNode.class);
 
+  public static final String CACHE_TYPE_LOCAL = "local";
+  public static final String CACHE_TYPE_MEMCACHED = "memcached";
+  public static final String CACHE_PROPERTY_PREFIX = "druid.bard.cache";
+
   private final List<Module> extraModules = Lists.newArrayList();
   private final List<String> pathsForGuiceFilter = Lists.newArrayList();
 
@@ -233,14 +240,31 @@ public class BrokerNode extends QueryableNode<BrokerNode>
   private void initializeCacheBroker()
   {
     if (cacheBroker == null) {
+      String cacheType = getConfigFactory()
+          .build(CacheConfig.class)
+          .getType();
+
+      if (cacheType.equals(CACHE_TYPE_LOCAL)) {
         setCacheBroker(
             MapCacheBroker.create(
                 getConfigFactory().buildWithReplacements(
                     MapCacheBrokerConfig.class,
-                  ImmutableMap.of("prefix", "druid.bard.cache")
+                    ImmutableMap.of("prefix", CACHE_PROPERTY_PREFIX)
                 )
             )
         );
+      } else if (cacheType.equals(CACHE_TYPE_MEMCACHED)) {
+        setCacheBroker(
+            MemcachedCacheBroker.create(
+                getConfigFactory().buildWithReplacements(
+                    MemcachedCacheBrokerConfig.class,
+                    ImmutableMap.of("prefix", CACHE_PROPERTY_PREFIX)
+                )
+            )
+        );
+      } else {
+        throw new ISE("Unknown cache type [%s]", cacheType);
+      }
     }
   }