Added stress test and cleaned up client request executor pool reset.

reset(dest)
- changed behavior to match original KeyedResourcePool implementation. The original QueuedKeyedResourcePool.reset() was an unnecessary/bad behavior change that canceled enqueued requests. The original behavior was to destroy idle resources whenever pool is reset.

test/integration/voldemort/socketpool/E2EClientRequestExecutorPoolAndFailureDetectorTest.java
- stress test that has put and get threads contend for slow servers in such a manner as to trigger failure detection to mark nodes unavailable. This excercises connection tear down, reset(), and build up again. This also exercises the code paths in which callbacks do heavyweight work.

src/java/voldemort/store/socket/clientrequest/ClientRequestExecutorPool.java

@@ -203,12 +203,7 @@ public void checkin(SocketDestination destination, ClientRequestExecutor clientR
     @Override
     public void close(SocketDestination destination) {
         factory.setLastClosedTimestamp(destination);
-        // TODO: Lazily destroy connections instead of actively destroying
-        // connections. Commenting out the next line changes to the lazy
-        // behavior. The other option is to change this behavior w/in
-        // QueuedKeyedResourcePool (i.e., instead of over-riding reset, just let
-        // KeyedResourcePool.reset be invoked directly).
-        // queuedPool.reset(destination);
+        queuedPool.reset(destination);
     }
 
     /**

src/java/voldemort/utils/pool/KeyedResourcePool.java

@@ -272,9 +272,9 @@ public void close() {
     }
 
     /**
-     * Reset a specific resource pool. Destroys all the resources in the pool.
-     * This method does not affect whether the pool is "open" in the sense of
-     * permitting new resources to be added to it.
+     * Reset a specific resource pool. Destroys all of the idle resources in the
+     * pool. This method does not affect whether the pool is "open" in the sense
+     * of permitting new resources to be added to it.
      * 
      * @param key The key for the pool to reset.
      */

src/java/voldemort/utils/pool/QueuedKeyedResourcePool.java

@@ -275,30 +275,6 @@ public void close() {
         internalClose();
     }
 
-    /**
-     * Reset a specific resource pool and resource request queue. First,
-     * "destroy" all registered resource requests. Second, destroy all resources
-     * in the pool.
-     * 
-     * @param key The key for the pool to reset.
-     */
-    @Override
-    public void reset(K key) {
-        // First, destroy enqueued resource requests (if any exist).
-        Queue<AsyncResourceRequest<V>> requestQueueToDestroy = requestQueueMap.get(key);
-        if(requestQueueToDestroy != null) {
-            // Swap in a new requestQueue so that the current requestQueue can
-            // be destroyed without the need of any synchronization primitives
-            Queue<AsyncResourceRequest<V>> newRequestQueue = new ConcurrentLinkedQueue<AsyncResourceRequest<V>>();
-            if(requestQueueMap.replace(key, requestQueueToDestroy, newRequestQueue)) {
-                destroyRequestQueue(requestQueueToDestroy);
-            }
-        }
-
-        // Second, destroy resources in the pool.
-        super.reset(key);
-    }
-
     /*
      * Get the queue of work for the given key. If no queue exists, create one.
      */