cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From alek...@apache.org
Subject cassandra git commit: Make randompartitioner work with new vnode allocation
Date Mon, 19 Sep 2016 23:38:15 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk f42e235b1 -> c1a9a47df


Make randompartitioner work with new vnode allocation

patch by Dikang Gu; reviewed by Branimir Lambov for CASSANDRA-12647


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/c1a9a47d
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/c1a9a47d
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/c1a9a47d

Branch: refs/heads/trunk
Commit: c1a9a47df292dbbde3c675c10d68043e7b212c28
Parents: f42e235
Author: Dikang Gu <dikang85@gmail.com>
Authored: Wed Sep 14 23:04:14 2016 -0700
Committer: Aleksey Yeschenko <aleksey@apache.org>
Committed: Mon Sep 19 16:36:34 2016 -0700

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../apache/cassandra/dht/RandomPartitioner.java | 13 +++
 .../ReplicationAwareTokenAllocatorTest.java     | 84 +++++++++++++++-----
 3 files changed, 76 insertions(+), 22 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/c1a9a47d/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 8e39d95..b625a58 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.10
+ * Make randompartitioner work with new vnode allocation (CASSANDRA-12647)
  * Fix cassandra-stress graphing (CASSANDRA-12237)
  * Allow filtering on partition key columns for queries without secondary indexes (CASSANDRA-11031)
  * Fix Cassandra Stress reporting thread model and precision (CASSANDRA-12585)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c1a9a47d/src/java/org/apache/cassandra/dht/RandomPartitioner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/dht/RandomPartitioner.java b/src/java/org/apache/cassandra/dht/RandomPartitioner.java
index c063be3..7c8f6ac 100644
--- a/src/java/org/apache/cassandra/dht/RandomPartitioner.java
+++ b/src/java/org/apache/cassandra/dht/RandomPartitioner.java
@@ -177,6 +177,19 @@ public class RandomPartitioner implements IPartitioner
         {
             return HEAP_SIZE;
         }
+
+        public Token increaseSlightly()
+        {
+            return new BigIntegerToken(token.add(BigInteger.ONE));
+        }
+
+        public double size(Token next)
+        {
+            BigIntegerToken n = (BigIntegerToken) next;
+            BigInteger v = n.token.subtract(token);  // Overflow acceptable and desired.
+            double d = Math.scalb(v.doubleValue(), -127); // Scale so that the full range
is 1.
+            return d > 0.0 ? d : (d + 1.0); // Adjust for signed long, also making sure
t.size(t) == 1.
+        }
     }
 
     public BigIntegerToken getToken(ByteBuffer key)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c1a9a47d/test/long/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocatorTest.java
----------------------------------------------------------------------
diff --git a/test/long/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocatorTest.java
b/test/long/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocatorTest.java
index 1b36c55..482e2ac 100644
--- a/test/long/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocatorTest.java
+++ b/test/long/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocatorTest.java
@@ -30,7 +30,9 @@ import org.apache.commons.math3.stat.descriptive.SummaryStatistics;
 import org.junit.Test;
 
 import org.apache.cassandra.Util;
+import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.RandomPartitioner;
 import org.apache.cassandra.dht.Token;
 
 public class ReplicationAwareTokenAllocatorTest
@@ -489,10 +491,10 @@ public class ReplicationAwareTokenAllocatorTest
         }
     };
 
-    Murmur3Partitioner partitioner = new Murmur3Partitioner();
     Random seededRand = new Random(2);
 
-    private void random(Map<Token, Unit> map, TestReplicationStrategy rs, int unitCount,
TokenCount tc, int perUnitCount)
+    private void random(Map<Token, Unit> map, TestReplicationStrategy rs,
+                        int unitCount, TokenCount tc, int perUnitCount, IPartitioner partitioner)
     {
         System.out.format("\nRandom generation of %d units with %d tokens each\n", unitCount,
perUnitCount);
         Random rand = seededRand;
@@ -509,49 +511,82 @@ public class ReplicationAwareTokenAllocatorTest
     }
 
     @Test
-    public void testExistingCluster()
+    public void testExistingClusterWithRandomPartitioner()
+    {
+        testExistingCluster(new RandomPartitioner());
+    }
+
+    @Test
+    public void testExistingClusterWithMurmur3Partitioner()
+    {
+        testExistingCluster(new Murmur3Partitioner());
+    }
+
+    public void testExistingCluster(IPartitioner partitioner)
     {
         for (int rf = 1; rf <= 5; ++rf)
         {
             for (int perUnitCount = 1; perUnitCount <= MAX_VNODE_COUNT; perUnitCount *=
4)
             {
-                testExistingCluster(perUnitCount, fixedTokenCount, new SimpleReplicationStrategy(rf));
-                testExistingCluster(perUnitCount, varyingTokenCount, new SimpleReplicationStrategy(rf));
+                testExistingCluster(perUnitCount, fixedTokenCount, new SimpleReplicationStrategy(rf),
partitioner);
+                testExistingCluster(perUnitCount, varyingTokenCount, new SimpleReplicationStrategy(rf),
partitioner);
                 if (rf == 1) continue;  // Replication strategy doesn't matter for RF = 1.
                 for (int groupSize = 4; groupSize <= 64 && groupSize * rf * 4
< TARGET_CLUSTER_SIZE; groupSize *= 4)
                 {
-                    testExistingCluster(perUnitCount, fixedTokenCount, new BalancedGroupReplicationStrategy(rf,
groupSize));
-                    testExistingCluster(perUnitCount, varyingTokenCount, new UnbalancedGroupReplicationStrategy(rf,
groupSize / 2, groupSize * 2, seededRand));
+                    testExistingCluster(perUnitCount, fixedTokenCount,
+                                        new BalancedGroupReplicationStrategy(rf, groupSize),
partitioner);
+                    testExistingCluster(perUnitCount, varyingTokenCount,
+                                        new UnbalancedGroupReplicationStrategy(rf, groupSize
/ 2, groupSize * 2, seededRand),
+                                        partitioner);
                 }
-                testExistingCluster(perUnitCount, fixedTokenCount, new FixedGroupCountReplicationStrategy(rf,
rf * 2));
+                testExistingCluster(perUnitCount, fixedTokenCount,
+                                    new FixedGroupCountReplicationStrategy(rf, rf * 2), partitioner);
             }
         }
     }
 
-    public void testExistingCluster(int perUnitCount, TokenCount tc, TestReplicationStrategy
rs)
+    public void testExistingCluster(int perUnitCount, TokenCount tc, TestReplicationStrategy
rs, IPartitioner partitioner)
     {
         System.out.println("Testing existing cluster, target " + perUnitCount + " vnodes,
replication " + rs);
         final int targetClusterSize = TARGET_CLUSTER_SIZE;
         NavigableMap<Token, Unit> tokenMap = Maps.newTreeMap();
 
-        random(tokenMap, rs, targetClusterSize / 2, tc, perUnitCount);
+        random(tokenMap, rs, targetClusterSize / 2, tc, perUnitCount, partitioner);
 
         ReplicationAwareTokenAllocator<Unit> t = new ReplicationAwareTokenAllocator<>(tokenMap,
rs, partitioner);
         grow(t, targetClusterSize * 9 / 10, tc, perUnitCount, false);
         grow(t, targetClusterSize, tc, perUnitCount, true);
-        loseAndReplace(t, targetClusterSize / 10, tc, perUnitCount);
+        loseAndReplace(t, targetClusterSize / 10, tc, perUnitCount, partitioner);
         System.out.println();
     }
 
     @Test
-    public void testNewCluster()
+    public void testNewClusterWithRandomPartitioner()
     {
-        Util.flakyTest(this::flakyTestNewCluster,
+        Util.flakyTest(this::flakyTestNewClusterWithRandomPartitioner,
                        5,
                        "It tends to fail sometimes due to the random selection of the tokens
in the first few nodes.");
     }
 
-    public void flakyTestNewCluster()
+    @Test
+    public void testNewClusterWithMurmur3Partitioner()
+    {
+        Util.flakyTest(this::flakyTestNewClusterWithMurmur3Partitioner,
+                       5,
+                       "It tends to fail sometimes due to the random selection of the tokens
in the first few nodes.");
+    }
+
+    public void flakyTestNewClusterWithRandomPartitioner()
+    {
+        flakyTestNewCluster(new RandomPartitioner());
+    }
+
+    public void flakyTestNewClusterWithMurmur3Partitioner()
+    {
+        flakyTestNewCluster(new Murmur3Partitioner());
+    }
+
+    public void flakyTestNewCluster(IPartitioner partitioner)
     {
         // This test is flaky because the selection of the tokens for the first RF nodes
(which is random, with an
         // uncontrolled seed) can sometimes cause a pathological situation where the algorithm
will find a (close to)
@@ -564,20 +599,24 @@ public class ReplicationAwareTokenAllocatorTest
         {
             for (int perUnitCount = 1; perUnitCount <= MAX_VNODE_COUNT; perUnitCount *=
4)
             {
-                testNewCluster(perUnitCount, fixedTokenCount, new SimpleReplicationStrategy(rf));
-                testNewCluster(perUnitCount, varyingTokenCount, new SimpleReplicationStrategy(rf));
+                testNewCluster(perUnitCount, fixedTokenCount, new SimpleReplicationStrategy(rf),
partitioner);
+                testNewCluster(perUnitCount, varyingTokenCount, new SimpleReplicationStrategy(rf),
partitioner);
                 if (rf == 1) continue;  // Replication strategy doesn't matter for RF = 1.
                 for (int groupSize = 4; groupSize <= 64 && groupSize * rf * 8
< TARGET_CLUSTER_SIZE; groupSize *= 4)
                 {
-                    testNewCluster(perUnitCount, fixedTokenCount, new BalancedGroupReplicationStrategy(rf,
groupSize));
-                    testNewCluster(perUnitCount, varyingTokenCount, new UnbalancedGroupReplicationStrategy(rf,
groupSize / 2, groupSize * 2, seededRand));
+                    testNewCluster(perUnitCount, fixedTokenCount,
+                                   new BalancedGroupReplicationStrategy(rf, groupSize), partitioner);
+                    testNewCluster(perUnitCount, varyingTokenCount,
+                                   new UnbalancedGroupReplicationStrategy(rf, groupSize /
2, groupSize * 2, seededRand),
+                                   partitioner);
                 }
-                testNewCluster(perUnitCount, fixedTokenCount, new FixedGroupCountReplicationStrategy(rf,
rf * 2));
+                testNewCluster(perUnitCount, fixedTokenCount,
+                               new FixedGroupCountReplicationStrategy(rf, rf * 2), partitioner);
             }
         }
     }
 
-    public void testNewCluster(int perUnitCount, TokenCount tc, TestReplicationStrategy rs)
+    public void testNewCluster(int perUnitCount, TokenCount tc, TestReplicationStrategy rs,
IPartitioner partitioner)
     {
         System.out.println("Testing new cluster, target " + perUnitCount + " vnodes, replication
" + rs);
         final int targetClusterSize = TARGET_CLUSTER_SIZE;
@@ -586,11 +625,12 @@ public class ReplicationAwareTokenAllocatorTest
         ReplicationAwareTokenAllocator<Unit> t = new ReplicationAwareTokenAllocator<>(tokenMap,
rs, partitioner);
         grow(t, targetClusterSize * 2 / 5, tc, perUnitCount, false);
         grow(t, targetClusterSize, tc, perUnitCount, true);
-        loseAndReplace(t, targetClusterSize / 5, tc, perUnitCount);
+        loseAndReplace(t, targetClusterSize / 5, tc, perUnitCount, partitioner);
         System.out.println();
     }
 
-    private void loseAndReplace(ReplicationAwareTokenAllocator<Unit> t, int howMany,
TokenCount tc, int perUnitCount)
+    private void loseAndReplace(ReplicationAwareTokenAllocator<Unit> t, int howMany,
+                                TokenCount tc, int perUnitCount, IPartitioner partitioner)
     {
         int fullCount = t.unitCount();
         System.out.format("Losing %d units. ", howMany);


Mime
View raw message