fix benchmarking code following removal of unnecessary method on MemcachedCacheBroker

client/src/test/java/com/metamx/druid/client/cache/MemcachedCacheBrokerBenchmark.java

@@ -3,6 +3,13 @@ package com.metamx.druid.client.cache;
 import com.google.caliper.Param;
 import com.google.caliper.Runner;
 import com.google.caliper.SimpleBenchmark;
+import net.spy.memcached.AddrUtil;
+import net.spy.memcached.ConnectionFactoryBuilder;
+import net.spy.memcached.DefaultHashAlgorithm;
+import net.spy.memcached.FailureMode;
+import net.spy.memcached.MemcachedClient;
+import net.spy.memcached.MemcachedClientIF;
+import net.spy.memcached.transcoders.SerializingTranscoder;
 
 import java.util.Random;
 import java.util.concurrent.TimeUnit;
@@ -12,6 +19,8 @@ public class MemcachedCacheBrokerBenchmark extends SimpleBenchmark
   private static final String BASE_KEY = "test_2012-11-26T00:00:00.000Z_2012-11-27T00:00:00.000Z_2012-11-27T04:11:25.979Z_";
 
   private MemcachedCacheBroker broker;
+  private MemcachedClientIF client;
+
   private Cache cache;
   private static byte[] randBytes;
 
@@ -22,36 +31,28 @@ public class MemcachedCacheBrokerBenchmark extends SimpleBenchmark
   @Override
   protected void setUp() throws Exception
   {
-    broker = MemcachedCacheBroker.create(
-        new MemcachedCacheBrokerConfig()
-        {
-          @Override
-          public int getExpiration()
-          {
-            // 1 year
-            return 3600 * 24 * 365;
-          }
+    SerializingTranscoder transcoder = new SerializingTranscoder(
+        50 * 1024 * 1024 // 50 MB
+    );
+    // disable compression
+    transcoder.setCompressionThreshold(Integer.MAX_VALUE);
 
-          @Override
-          public int getTimeout()
-          {
-            // 500 milliseconds
-            return 500;
-          }
+    client = new MemcachedClient(
+        new ConnectionFactoryBuilder().setProtocol(ConnectionFactoryBuilder.Protocol.BINARY)
+                                      .setHashAlg(DefaultHashAlgorithm.FNV1A_64_HASH)
+                                      .setLocatorType(ConnectionFactoryBuilder.Locator.CONSISTENT)
+                                      .setDaemon(true)
+                                      .setFailureMode(FailureMode.Retry)
+                                      .setTranscoder(transcoder)
+                                      .setShouldOptimize(true)
+                                      .build(),
+        AddrUtil.getAddresses("localhost:11211")
+    );
 
-          @Override
-          public String getHosts()
-          {
-            return "localhost:11211";
-          }
-
-          @Override
-          public int getMaxObjectSize()
-          {
-            // 50 MB
-            return 50 * 1024 * 1024;
-          }
-        }
+    broker = new MemcachedCacheBroker(
+        client,
+        500, // 500 milliseconds
+        3600 * 24 * 365 // 1 year
     );
 
     cache = broker.provideCache("default");
@@ -64,8 +65,8 @@ public class MemcachedCacheBrokerBenchmark extends SimpleBenchmark
   @Override
   protected void tearDown() throws Exception
   {
-    broker.getClient().flush();
-    broker.getClient().shutdown();
+    client.flush();
+    client.shutdown();
   }
 
   public void timePutObjects(int reps) {
@@ -74,7 +75,8 @@ public class MemcachedCacheBrokerBenchmark extends SimpleBenchmark
         String key = BASE_KEY + i;
         cache.put(key.getBytes(), randBytes);
       }
-      broker.getClient().waitForQueues(1, TimeUnit.HOURS);
+      // make sure the write queue is empty
+      client.waitForQueues(1, TimeUnit.HOURS);
     }
   }