Added unit tests to ensure that slops are registered for different as…

…ynchronous put operation failures

src/java/voldemort/store/routed/PipelineRoutedStore.java

@@ -29,11 +29,11 @@
 import voldemort.cluster.failuredetector.FailureDetector;
 import voldemort.common.VoldemortOpCode;
 import voldemort.routing.RoutingStrategyType;
+import voldemort.store.CompositeVoldemortRequest;
 import voldemort.store.Store;
 import voldemort.store.StoreDefinition;
 import voldemort.store.StoreRequest;
 import voldemort.store.StoreUtils;
-import voldemort.store.CompositeVoldemortRequest;
 import voldemort.store.nonblockingstore.NonblockingStore;
 import voldemort.store.routed.Pipeline.Event;
 import voldemort.store.routed.Pipeline.Operation;
@@ -75,10 +75,10 @@
  */
 public class PipelineRoutedStore extends RoutedStore {
 
-    private final Map<Integer, NonblockingStore> nonblockingStores;
-    private final Map<Integer, Store<ByteArray, Slop, byte[]>> slopStores;
-    private final Map<Integer, NonblockingStore> nonblockingSlopStores;
-    private final HintedHandoffStrategy handoffStrategy;
+    protected final Map<Integer, NonblockingStore> nonblockingStores;
+    protected final Map<Integer, Store<ByteArray, Slop, byte[]>> slopStores;
+    protected final Map<Integer, NonblockingStore> nonblockingSlopStores;
+    protected final HintedHandoffStrategy handoffStrategy;
     private Zone clientZone;
     private boolean zoneRoutingEnabled;
     private PipelineRoutedStats stats;

test/common/voldemort/store/SleepyForceFailStore.java

@@ -0,0 +1,60 @@
+package voldemort.store;
+
+import java.util.List;
+import java.util.Map;
+
+import voldemort.VoldemortException;
+import voldemort.versioning.Version;
+import voldemort.versioning.Versioned;
+
+public class SleepyForceFailStore<K, V, T> extends ForceFailStore<K, V, T> {
+
+    private long sleepTimeMs;
+
+    public SleepyForceFailStore(Store<K, V, T> innerStore, VoldemortException e, long sleepTimeInMs) {
+        super(innerStore, e);
+        this.sleepTimeMs = sleepTimeInMs;
+    }
+
+    @Override
+    public boolean delete(K key, Version version) throws VoldemortException {
+        try {
+            Thread.sleep(sleepTimeMs);
+            return super.delete(key, version);
+        } catch(InterruptedException e) {
+            throw new VoldemortException(e);
+        }
+    }
+
+    @Override
+    public List<Versioned<V>> get(K key, T transforms) throws VoldemortException {
+        try {
+            Thread.sleep(sleepTimeMs);
+            return super.get(key, transforms);
+        } catch(InterruptedException e) {
+            throw new VoldemortException(e);
+        }
+    }
+
+    @Override
+    public Map<K, List<Versioned<V>>> getAll(Iterable<K> keys, Map<K, T> transforms)
+            throws VoldemortException {
+        try {
+            Thread.sleep(sleepTimeMs);
+            return super.getAll(keys, transforms);
+        } catch(InterruptedException e) {
+            throw new VoldemortException(e);
+        }
+    }
+
+    @Override
+    public void put(K key, Versioned<V> value, T transforms) throws VoldemortException {
+        try {
+            Thread.sleep(sleepTimeMs);
+            super.put(key, value, transforms);
+        } catch(InterruptedException e) {
+            throw new VoldemortException(e);
+        }
+    }
+
+}