Merge branch 'cassandra-2.2' into cassandra-3.0

# Conflicts:
#	CHANGES.txt
#	src/java/org/apache/cassandra/locator/TokenMetadata.java

CHANGES.txt

@@ -1,4 +1,6 @@
 3.0.25:
+Merged from 2.2:
+ * Make TokenMetadata's ring version increments atomic (CASSANDRA-16286)
 
 
 3.0.24:

src/java/org/apache/cassandra/locator/TokenMetadata.java

@@ -26,6 +26,8 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import javax.annotation.concurrent.GuardedBy;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.*;
 import org.apache.commons.lang3.StringUtils;
@@ -110,7 +112,8 @@ public class TokenMetadata
     };
 
     // signals replication strategies that nodes have joined or left the ring and they need to recompute ownership
-    private volatile long ringVersion = 0;
+    @GuardedBy("lock")
+    private long ringVersion = 0;
 
     public TokenMetadata()
     {
@@ -476,7 +479,7 @@ public class TokenMetadata
             }
             endpointToHostIdMap.remove(endpoint);
             sortedTokens = sortTokens();
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
         }
         finally
         {
@@ -496,7 +499,7 @@ public class TokenMetadata
         {
             logger.info("Updating topology for {}", endpoint);
             topology = topology.unbuild().updateEndpoint(endpoint).build();
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
             return topology;
         }
         finally
@@ -516,7 +519,7 @@ public class TokenMetadata
         {
             logger.info("Updating topology for all endpoints that have changed");
             topology = topology.unbuild().updateEndpoints().build();
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
             return topology;
         }
         finally
@@ -545,7 +548,7 @@ public class TokenMetadata
                 }
             }
 
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
         }
         finally
         {
@@ -1134,7 +1137,7 @@ public class TokenMetadata
             movingEndpoints.clear();
             sortedTokens.clear();
             topology = Topology.empty();
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
         }
         finally
         {
@@ -1280,10 +1283,33 @@ public class TokenMetadata
 
     public long getRingVersion()
     {
-        return ringVersion;
+        lock.readLock().lock();
+
+        try
+        {
+            return ringVersion;
+        }
+        finally
+        {
+            lock.readLock().unlock();
+        }
     }
 
     public void invalidateCachedRings()
+    {   
+        lock.writeLock().lock();
+
+        try
+        {   
+            invalidateCachedRingsUnsafe();
+        }
+        finally
+        {
+            lock.writeLock().unlock();
+        }
+    }
+    
+    private void invalidateCachedRingsUnsafe()
     {
         ringVersion++;
         cachedTokenMap.set(null);

test/unit/org/apache/cassandra/locator/TokenMetadataTest.java

@@ -21,6 +21,9 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.Iterators;
@@ -68,6 +71,27 @@ public class TokenMetadataTest
             assertEquals("Mismatch at index " + i + ": " + actual, token(expected[i]), actual.get(i));
     }
 
+    /**
+     * This test is very likely (but not guaranteed) to fail if ring invalidations are ever allowed to interleave.
+     */
+    @Test
+    public void testConcurrentInvalidation() throws InterruptedException
+    {
+        long startVersion = tmd.getRingVersion();
+
+        ExecutorService pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() + 1);
+        
+        int invalidations = 1024;
+        
+        for (int i = 0; i < invalidations; i++)
+            pool.execute(() -> tmd.invalidateCachedRings());
+
+        pool.shutdown();
+        
+        assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS));
+        assertEquals(invalidations + startVersion, tmd.getRingVersion());
+    }
+
     @Test
     public void testRingIterator()
     {