Fixed BulkGetFuture.get() API to honor operationTimeout.

src/main/java/net/spy/memcached/MemcachedClient.java

@@ -1319,7 +1319,7 @@ public <T> BulkFuture<Map<String, T>> asyncGetBulk(Iterator<String> keyIter,
     int initialLatchCount = chunks.isEmpty() ? 0 : 1;
     final CountDownLatch latch = new CountDownLatch(initialLatchCount);
     final Collection<Operation> ops = new ArrayList<Operation>(chunks.size());
-    final BulkGetFuture<T> rv = new BulkGetFuture<T>(m, ops, latch, executorService);
+    final BulkGetFuture<T> rv = new BulkGetFuture<T>(m, ops, latch, operationTimeout, executorService);
 
     GetOperation.Callback cb = new GetOperation.Callback() {
       @Override

src/main/java/net/spy/memcached/internal/BulkGetFuture.java

@@ -56,16 +56,18 @@ public class BulkGetFuture<T>
   private final Map<String, Future<T>> rvMap;
   private final Collection<Operation> ops;
   private final CountDownLatch latch;
+  private final long defaultTimeoutMillis;
   private OperationStatus status;
   private boolean cancelled = false;
   private boolean timeout = false;
 
   public BulkGetFuture(Map<String, Future<T>> m, Collection<Operation> getOps,
-      CountDownLatch l, ExecutorService service) {
+      CountDownLatch l, long defaultTimeoutMillis, ExecutorService service) {
     super(service);
     rvMap = m;
     ops = getOps;
     latch = l;
+    this.defaultTimeoutMillis = defaultTimeoutMillis;
     status = null;
   }
 
@@ -86,9 +88,9 @@ public boolean cancel(boolean ign) {
 
   public Map<String, T> get() throws InterruptedException, ExecutionException {
     try {
-      return get(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
+      return get(defaultTimeoutMillis, TimeUnit.MILLISECONDS);
     } catch (TimeoutException e) {
-      throw new RuntimeException("Timed out waiting forever", e);
+      throw new ExecutionException("Bulk operation timed out after " + defaultTimeoutMillis + " millis", e);
     }
   }
 

src/test/java/net/spy/memcached/TimeoutTest.java

@@ -23,6 +23,10 @@
 
 package net.spy.memcached;
 
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
 /**
  * A TimeoutTest.
  */
@@ -90,6 +94,39 @@ public void run() {
     });
   }
 
+  public void testAsyncGetBulkCustomTimeout() {
+    tryTimeout("asyncGetBulk", new Runnable() {
+      public void run() {
+        try {
+          client.asyncGetBulk("k", "k2").get(500, TimeUnit.MILLISECONDS);
+        } catch (TimeoutException e) {
+          throw new OperationTimeoutException("Bulk op timed out - custom timeout", e);
+        } catch (Exception e) {
+          throw new RuntimeException("Unexpected exception in bulk op", e);
+        }
+      }
+    });
+  }
+
+  public void testAsyncGetBulkDefaultTimeout() {
+    tryTimeout("asyncGetBulk", new Runnable() {
+      public void run() {
+        try {
+          client.asyncGetBulk("k", "k2").get();
+        } catch (ExecutionException e) {
+          if (e.getCause() instanceof TimeoutException) {
+            throw new OperationTimeoutException("Bulk op timed out - default timeout", e.getCause());
+          }
+          else {
+            throw new RuntimeException("Unexpected execution exception in bulk op", e);
+          }
+        } catch (Exception e) {
+          throw new RuntimeException("Unexpected exception in bulk op", e);
+        }
+      }
+    });
+  }
+
   public void testIncrTimeout() {
     tryTimeout("incr", new Runnable() {
       public void run() {