Added test and patch for RoutedStore.put timeout issue

src/java/voldemort/store/routed/RoutedStore.java

@@ -51,6 +51,7 @@
 import voldemort.store.StoreUtils;
 import voldemort.store.UnreachableStoreException;
 import voldemort.utils.ByteArray;
+import voldemort.utils.ByteUtils;
 import voldemort.utils.SystemTime;
 import voldemort.utils.Time;
 import voldemort.utils.Utils;
@@ -222,7 +223,7 @@ public void run() {
                     boolean acquired = semaphore.tryAcquire(timeoutMs, TimeUnit.MILLISECONDS);
                     if(!acquired)
                         logger.warn("Delete operation timed out waiting for operation " + i
-                                    + "to complete after waiting " + timeoutMs + " ms.");
+                                    + " to complete after waiting " + timeoutMs + " ms.");
                     // okay, at least the required number of operations have
                     // completed, were they successful?
                     if(successes.get() >= attempts)
@@ -450,6 +451,9 @@ public List<Versioned<byte[]>> get(final ByteArray key) throws VoldemortExceptio
 
                 public void run() {
                     try {
+                        if(logger.isTraceEnabled())
+                            logger.trace("Attempting get operation on node " + node.getId()
+                                         + " for key '" + ByteUtils.toHexString(key.get()) + "'.");
                         List<Versioned<byte[]>> fetched = innerStores.get(node.getId()).get(key);
                         retrieved.addAll(fetched);
                         if(repairReads) {
@@ -577,6 +581,7 @@ public String getName() {
 
     public void put(final ByteArray key, final Versioned<byte[]> versioned)
             throws VoldemortException {
+        long start = System.currentTimeMillis();
         StoreUtils.assertValidKey(key);
         final List<Node> nodes = availableNodes(routingStrategy.routeRequest(key.get()));
 
@@ -665,9 +670,13 @@ public void run() {
         int blockCount = Math.min(storeDef.getPreferredWrites() - 1, attempts);
         for(int i = 0; i < blockCount; i++) {
             try {
-                boolean acquired = semaphore.tryAcquire(timeoutMs, TimeUnit.MILLISECONDS);
-                if(!acquired)
+                long delta = System.currentTimeMillis() - start;
+                long timeout = timeoutMs - delta;
+                boolean acquired = (timeout > 0) ? semaphore.tryAcquire(timeout, TimeUnit.MILLISECONDS) : false;
+                if(!acquired) {
                     logger.warn("Timed out waiting for put to succeed.");
+                    break;
+                }
                 // okay, at least the required number of operations have
                 // completed, were they successful?
                 if(successes.get() >= this.storeDef.getPreferredWrites())

test/unit/voldemort/store/routed/TimeoutTest.java

@@ -0,0 +1,120 @@
+package voldemort.store.routed;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import junit.framework.TestCase;
+import voldemort.VoldemortException;
+import voldemort.client.RoutingTier;
+import voldemort.cluster.Cluster;
+import voldemort.cluster.Node;
+import voldemort.routing.RoutingStrategyType;
+import voldemort.serialization.SerializerDefinition;
+import voldemort.store.InsufficientOperationalNodesException;
+import voldemort.store.Store;
+import voldemort.store.StoreCapabilityType;
+import voldemort.store.StoreDefinition;
+import voldemort.utils.ByteArray;
+import voldemort.versioning.Version;
+import voldemort.versioning.Versioned;
+
+/**
+ * 
+ * Test the configured timeout operate as expected.
+ * 
+ * @author jtuberville
+ * 
+ */
+public class TimeoutTest extends TestCase {
+
+    class SlowStore implements Store<ByteArray, byte[]> {
+
+        private final long delay;
+
+        SlowStore(long delay) {
+            this.delay = delay;
+        }
+
+        public void close() throws VoldemortException {
+            throw new UnsupportedOperationException();
+        }
+
+        public boolean delete(ByteArray key, Version version) throws VoldemortException {
+            throw new UnsupportedOperationException();
+        }
+
+        public List<Versioned<byte[]>> get(ByteArray key) throws VoldemortException {
+            throw new UnsupportedOperationException();
+        }
+
+        public Map<ByteArray, List<Versioned<byte[]>>> getAll(Iterable<ByteArray> keys)
+                throws VoldemortException {
+            throw new UnsupportedOperationException();
+        }
+
+        public Object getCapability(StoreCapabilityType capability) {
+            throw new UnsupportedOperationException();
+        }
+
+        public String getName() {
+            throw new UnsupportedOperationException();
+        }
+
+        public void put(ByteArray key, Versioned<byte[]> value) throws VoldemortException {
+            try {
+                Thread.sleep(delay);
+            } catch(InterruptedException e) {
+                throw new VoldemortException(e);
+            }
+        }
+
+    }
+
+    public void testPutTimeout() {
+        int timeout = 50;
+        StoreDefinition definition = new StoreDefinition("test",
+                                                         "foo",
+                                                         new SerializerDefinition("test"),
+                                                         new SerializerDefinition("test"),
+                                                         RoutingTier.CLIENT,
+                                                         RoutingStrategyType.CONSISTENT_STRATEGY,
+                                                         3,
+                                                         3,
+                                                         3,
+                                                         3,
+                                                         3,
+                                                         0);
+        Map<Integer, Store<ByteArray, byte[]>> stores = new HashMap<Integer, Store<ByteArray, byte[]>>();
+        List<Node> nodes = new ArrayList<Node>();
+        int totalDelay = 0;
+        for(int i = 0; i < 3; i++) {
+            int delay = 4 + i * timeout;
+            totalDelay += delay;
+            stores.put(i, new SlowStore(delay));
+            List<Integer> partitions = Arrays.asList(i);
+            nodes.add(new Node(i, "none", 0, 0, partitions));
+        }
+
+        RoutedStore routedStore = new RoutedStore("test",
+                                                  stores,
+                                                  new Cluster("test", nodes),
+                                                  definition,
+                                                  3,
+                                                  false,
+                                                  timeout);
+
+        long start = System.currentTimeMillis();
+        try {
+            routedStore.put(new ByteArray("test".getBytes()),
+                            new Versioned<byte[]>(new byte[] { 1 }));
+            fail("Should have thrown");
+        } catch(InsufficientOperationalNodesException e) {
+            long elapsed = System.currentTimeMillis() - start;
+            assertTrue(elapsed + " < " + totalDelay, elapsed < totalDelay);
+        }
+
+    }
+}