SPY-160: Avoid ConcurrentModificationException with the retry queue

Motivation
----------
ConcurrentModificationException will show up since the code can recursively
modify itself.

Modifications
-------------
Copy the original list into a temporary new one to avoid the exception.

Result
------
Redistribute on the retry list will work as expected, without an
exception.

Change-Id: Ieff1a87136bef38280b7f139192419073d6a878a
Reviewed-on: http://review.couchbase.org/34977
Tested-by: Michael Nitschinger <michael.nitschinger@couchbase.com>
Reviewed-by: Matt Ingenthron <matt@couchbase.com>

Author: daschl

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

@@ -38,6 +38,7 @@
 import java.text.MessageFormat;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.ConcurrentModificationException;
 import java.util.HashSet;
 import java.util.IdentityHashMap;
@@ -196,7 +197,7 @@ public class MemcachedConnection extends SpyThread {
   /**
    * Holds operations that need to be retried.
    */
-  private final Collection<Operation> retryOps;
+  private final List<Operation> retryOps;
 
   /**
    * Holds all nodes that are scheduled for shutdown.
@@ -247,7 +248,7 @@ public MemcachedConnection(final int bufSize, final ConnectionFactory f,
     opFact = opfactory;
     timeoutExceptionThreshold = f.getTimeoutExceptionThreshold();
     selector = Selector.open();
-    retryOps = new ArrayList<Operation>();
+    retryOps = Collections.synchronizedList(new ArrayList<Operation>());
     nodesToShutdown = new ConcurrentLinkedQueue<MemcachedNode>();
     listenerExecutorService = f.getListenerExecutorService();
     this.bufSize = bufSize;
@@ -429,8 +430,11 @@ private void handleOperationalTasks() throws IOException {
     if (!shutDown && !reconnectQueue.isEmpty()) {
       attemptReconnects();
     }
-    redistributeOperations(retryOps);
-    retryOps.clear();
+
+    if (!retryOps.isEmpty()) {
+      redistributeOperations(new ArrayList<Operation>(retryOps));
+      retryOps.clear();
+    }
 
     handleShutdownQueue();
   }