Merge branch 'cassandra-2.0' into cassandra-2.1

Conflicts:
	CHANGES.txt
	test/unit/org/apache/cassandra/service/StorageServiceServerTest.java

CHANGES.txt

@@ -84,6 +84,7 @@ Merged from 2.0:
  * cqlsh fails when version number parts are not int (CASSANDRA-7524)
  * Fix NPE when table dropped during streaming (CASSANDRA-7946)
  * Fix wrong progress when streaming uncompressed (CASSANDRA-7878)
+ * Fix possible infinite loop in creating repair range (CASSANDRA-7983)
 Merged from 1.2:
  * Don't index tombstones (CASSANDRA-7828)
  * Improve PasswordAuthenticator default super user setup (CASSANDRA-7788)

src/java/org/apache/cassandra/service/StorageService.java

@@ -43,6 +43,7 @@ import ch.qos.logback.classic.jmx.JMXConfiguratorMBean;
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.Appender;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Predicate;
 import com.google.common.collect.*;
 import com.google.common.util.concurrent.FutureCallback;
@@ -2579,22 +2580,33 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
      * @return collection of ranges that match ring layout in TokenMetadata
      */
     @SuppressWarnings("unchecked")
-    private Collection<Range<Token>> createRepairRangeFrom(String beginToken, String endToken)
+    @VisibleForTesting
+    Collection<Range<Token>> createRepairRangeFrom(String beginToken, String endToken)
     {
         Token parsedBeginToken = getPartitioner().getTokenFactory().fromString(beginToken);
         Token parsedEndToken = getPartitioner().getTokenFactory().fromString(endToken);
 
-        Deque<Range<Token>> repairingRange = new ArrayDeque<>();
         // Break up given range to match ring layout in TokenMetadata
-        Token previous = tokenMetadata.getPredecessor(TokenMetadata.firstToken(tokenMetadata.sortedTokens(), parsedEndToken));
+        ArrayList<Range<Token>> repairingRange = new ArrayList<>();
-        while (parsedBeginToken.compareTo(previous) < 0)
-        {
-            repairingRange.addFirst(new Range<>(previous, parsedEndToken));
 
-            parsedEndToken = previous;
+        ArrayList<Token> tokens = new ArrayList<>(tokenMetadata.sortedTokens());
-            previous = tokenMetadata.getPredecessor(previous);
+        if (!tokens.contains(parsedBeginToken))
+        {
+            tokens.add(parsedBeginToken);
+        }
+        if (!tokens.contains(parsedEndToken))
+        {
+            tokens.add(parsedEndToken);
+        }
+        // tokens now contain all tokens including our endpoints
+        Collections.sort(tokens);
+
+        int start = tokens.indexOf(parsedBeginToken), end = tokens.indexOf(parsedEndToken);
+        for (int i = start; i != end; i = (i+1) % tokens.size())
+        {
+            Range<Token> range = new Range<>(tokens.get(i), tokens.get((i+1) % tokens.size()));
+            repairingRange.add(range);
         }
-        repairingRange.addFirst(new Range<>(parsedBeginToken, parsedEndToken));
 
         return repairingRange;
     }

test/unit/org/apache/cassandra/service/StorageServiceServerTest.java

@@ -26,6 +26,10 @@ import java.util.*;
 
 import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Multimap;
+
+import org.apache.cassandra.dht.BigIntegerToken;
+import org.apache.cassandra.dht.LongToken;
+import org.apache.cassandra.dht.Murmur3Partitioner;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -458,4 +462,50 @@ public class StorageServiceServerTest
         assert primaryRanges.size() == 1;
         assert primaryRanges.contains(new Range<Token>(new StringToken("B"), new StringToken("C")));
     }
+
+    @Test
+    public void testCreateRepairRangeFrom() throws Exception
+    {
+        StorageService.instance.setPartitionerUnsafe(new Murmur3Partitioner());
+
+        TokenMetadata metadata = StorageService.instance.getTokenMetadata();
+        metadata.clearUnsafe();
+
+        metadata.updateNormalToken(new LongToken(1000L), InetAddress.getByName("127.0.0.1"));
+        metadata.updateNormalToken(new LongToken(2000L), InetAddress.getByName("127.0.0.2"));
+        metadata.updateNormalToken(new LongToken(3000L), InetAddress.getByName("127.0.0.3"));
+        metadata.updateNormalToken(new LongToken(4000L), InetAddress.getByName("127.0.0.4"));
+
+        Map<String, String> configOptions = new HashMap<String, String>();
+        configOptions.put("replication_factor", "3");
+        Collection<Range<Token>> repairRangeFrom = StorageService.instance.createRepairRangeFrom("1500", "3700");
+        assert repairRangeFrom.size() == 3;
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(1500L), new LongToken(2000L)));
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(2000L), new LongToken(3000L)));
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(3000L), new LongToken(3700L)));
+
+        repairRangeFrom = StorageService.instance.createRepairRangeFrom("500", "700");
+        assert repairRangeFrom.size() == 1;
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(500L), new LongToken(700L)));
+
+        repairRangeFrom = StorageService.instance.createRepairRangeFrom("500", "1700");
+        assert repairRangeFrom.size() == 2;
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(500L), new LongToken(1000L)));
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(1000L), new LongToken(1700L)));
+
+        repairRangeFrom = StorageService.instance.createRepairRangeFrom("2500", "2300");
+        assert repairRangeFrom.size() == 5;
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(2500L), new LongToken(3000L)));
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(3000L), new LongToken(4000L)));
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(4000L), new LongToken(1000L)));
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(1000L), new LongToken(2000L)));
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(2000L), new LongToken(2300L)));
+
+        repairRangeFrom = StorageService.instance.createRepairRangeFrom("2000", "3000");
+        assert repairRangeFrom.size() == 1;
+        assert repairRangeFrom.contains(new Range<Token>(new LongToken(2000L), new LongToken(3000L)));
+
+        repairRangeFrom = StorageService.instance.createRepairRangeFrom("2000", "2000");
+        assert repairRangeFrom.size() == 0;
+    }
 }