Repository: cassandra
Updated Branches:
  refs/heads/cassandra-3.X 7a5118c7d -> c6ec31bad
  refs/heads/trunk f2a354763 -> a4705c28b


Fix timeout in ReplicationAwareTokenAllocatorTest

patch by Stefania Alborghetti; reviewed by Branimir Lambov for CASSANDRA-12784


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

Branch: refs/heads/cassandra-3.X
Commit: c6ec31bada33b9b803d09a863414ea9cad72752e
Parents: 7a5118c
Author: Stefania Alborghetti <stefania.alborghe...@datastax.com>
Authored: Thu Oct 13 16:32:58 2016 +0800
Committer: Stefania Alborghetti <stefania.alborghe...@datastax.com>
Committed: Tue Oct 18 09:08:56 2016 +0800

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../apache/cassandra/dht/RandomPartitioner.java |   2 +-
 .../apache/cassandra/utils/GuidGenerator.java   |   5 +-
 ...tractReplicationAwareTokenAllocatorTest.java | 716 ++++++++++++++++++
 ...rmur3ReplicationAwareTokenAllocatorTest.java |  48 ++
 ...andomReplicationAwareTokenAllocatorTest.java |  54 ++
 .../ReplicationAwareTokenAllocatorTest.java     | 755 -------------------
 test/unit/org/apache/cassandra/Util.java        |  20 +-
 8 files changed, 835 insertions(+), 766 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/c6ec31ba/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index d230462..32a2dfd 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.10
+ * Fix timeout in ReplicationAwareTokenAllocatorTest (CASSANDRA-12784)
  * Improve sum aggregate functions (CASSANDRA-12417)
  * Make cassandra.yaml docs for batch_size_*_threshold_in_kb reflect changes 
in CASSANDRA-10876 (CASSANDRA-12761)
  * cqlsh fails to format collections when using aliases (CASSANDRA-11534)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c6ec31ba/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 7c8f6ac..ee3b862 100644
--- a/src/java/org/apache/cassandra/dht/RandomPartitioner.java
+++ b/src/java/org/apache/cassandra/dht/RandomPartitioner.java
@@ -93,7 +93,7 @@ public class RandomPartitioner implements IPartitioner
 
     public BigIntegerToken getRandomToken(Random random)
     {
-        BigInteger token = 
FBUtilities.hashToBigInteger(GuidGenerator.guidAsBytes(random));
+        BigInteger token = 
FBUtilities.hashToBigInteger(GuidGenerator.guidAsBytes(random, 0));
         if ( token.signum() == -1 )
             token = token.multiply(BigInteger.valueOf(-1L));
         return new BigIntegerToken(token);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c6ec31ba/src/java/org/apache/cassandra/utils/GuidGenerator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/GuidGenerator.java 
b/src/java/org/apache/cassandra/utils/GuidGenerator.java
index 0843344..c5ed7a7 100644
--- a/src/java/org/apache/cassandra/utils/GuidGenerator.java
+++ b/src/java/org/apache/cassandra/utils/GuidGenerator.java
@@ -76,10 +76,9 @@ public class GuidGenerator
         return convertToStandardFormat( sb.toString() );
     }
 
-    public static ByteBuffer guidAsBytes(Random random)
+    public static ByteBuffer guidAsBytes(Random random, long time)
     {
         StringBuilder sbValueBeforeMD5 = new StringBuilder();
-        long time = System.currentTimeMillis();
         long rand = 0;
         rand = random.nextLong();
         sbValueBeforeMD5.append(s_id)
@@ -94,7 +93,7 @@ public class GuidGenerator
 
     public static ByteBuffer guidAsBytes()
     {
-        return guidAsBytes(myRand);
+        return guidAsBytes(myRand, System.currentTimeMillis());
     }
 
     /*

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c6ec31ba/test/long/org/apache/cassandra/dht/tokenallocator/AbstractReplicationAwareTokenAllocatorTest.java
----------------------------------------------------------------------
diff --git 
a/test/long/org/apache/cassandra/dht/tokenallocator/AbstractReplicationAwareTokenAllocatorTest.java
 
b/test/long/org/apache/cassandra/dht/tokenallocator/AbstractReplicationAwareTokenAllocatorTest.java
new file mode 100644
index 0000000..80e980a
--- /dev/null
+++ 
b/test/long/org/apache/cassandra/dht/tokenallocator/AbstractReplicationAwareTokenAllocatorTest.java
@@ -0,0 +1,716 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.dht.tokenallocator;
+
+import java.util.*;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import org.apache.commons.math3.stat.descriptive.SummaryStatistics;
+
+import org.junit.Assert;
+import org.junit.Ignore;
+
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Token;
+
+/**
+ * Base class for {@link Murmur3ReplicationAwareTokenAllocatorTest} and {@link 
RandomReplicationAwareTokenAllocatorTest},
+ * we need to separate classes to avoid timeous in case flaky tests need to be 
repeated, see CASSANDRA-12784.
+ */
+@Ignore
+abstract class AbstractReplicationAwareTokenAllocatorTest
+{
+    private static final int TARGET_CLUSTER_SIZE = 250;
+
+    interface TestReplicationStrategy extends ReplicationStrategy<Unit>
+    {
+        void addUnit(Unit n);
+
+        void removeUnit(Unit n);
+
+        /**
+         * Returns a list of all replica units for given token.
+         */
+        List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> 
sortedTokens);
+
+        /**
+         * Returns the start of the token span that is replicated in this 
token.
+         * Note: Though this is not trivial to see, the replicated span is 
always contiguous. A token in the same
+         * group acts as a barrier; if one is not found the token replicates 
everything up to the replica'th distinct
+         * group seen in front of it.
+         */
+        Token replicationStart(Token token, Unit unit, NavigableMap<Token, 
Unit> sortedTokens);
+
+        /**
+         * Multiplier for the acceptable disbalance in the cluster. With some 
strategies it is harder to achieve good
+         * results.
+         */
+        public double spreadExpectation();
+    }
+
+    static class NoReplicationStrategy implements TestReplicationStrategy
+    {
+        public List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> 
sortedTokens)
+        {
+            return 
Collections.singletonList(sortedTokens.ceilingEntry(token).getValue());
+        }
+
+        public Token replicationStart(Token token, Unit unit, 
NavigableMap<Token, Unit> sortedTokens)
+        {
+            return sortedTokens.lowerKey(token);
+        }
+
+        public String toString()
+        {
+            return "No replication";
+        }
+
+        public void addUnit(Unit n)
+        {
+        }
+
+        public void removeUnit(Unit n)
+        {
+        }
+
+        public int replicas()
+        {
+            return 1;
+        }
+
+        public boolean sameGroup(Unit n1, Unit n2)
+        {
+            return false;
+        }
+
+        public Object getGroup(Unit unit)
+        {
+            return unit;
+        }
+
+        public double spreadExpectation()
+        {
+            return 1;
+        }
+    }
+
+    static class SimpleReplicationStrategy implements TestReplicationStrategy
+    {
+        int replicas;
+
+        public SimpleReplicationStrategy(int replicas)
+        {
+            super();
+            this.replicas = replicas;
+        }
+
+        public List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> 
sortedTokens)
+        {
+            List<Unit> endpoints = new ArrayList<Unit>(replicas);
+
+            token = sortedTokens.ceilingKey(token);
+            if (token == null)
+                token = sortedTokens.firstKey();
+            Iterator<Unit> iter = Iterables.concat(sortedTokens.tailMap(token, 
true).values(), sortedTokens.values()).iterator();
+            while (endpoints.size() < replicas)
+            {
+                if (!iter.hasNext())
+                    return endpoints;
+                Unit ep = iter.next();
+                if (!endpoints.contains(ep))
+                    endpoints.add(ep);
+            }
+            return endpoints;
+        }
+
+        public Token replicationStart(Token token, Unit unit, 
NavigableMap<Token, Unit> sortedTokens)
+        {
+            Set<Unit> seenUnits = Sets.newHashSet();
+            int unitsFound = 0;
+
+            for (Map.Entry<Token, Unit> en : Iterables.concat(
+                                                             
sortedTokens.headMap(token, false).descendingMap().entrySet(),
+                                                             
sortedTokens.descendingMap().entrySet()))
+            {
+                Unit n = en.getValue();
+                // Same group as investigated unit is a break; anything that 
could replicate in it replicates there.
+                if (n == unit)
+                    break;
+
+                if (seenUnits.add(n))
+                {
+                    if (++unitsFound == replicas)
+                        break;
+                }
+                token = en.getKey();
+            }
+            return token;
+        }
+
+        public void addUnit(Unit n)
+        {
+        }
+
+        public void removeUnit(Unit n)
+        {
+        }
+
+        public String toString()
+        {
+            return String.format("Simple %d replicas", replicas);
+        }
+
+        public int replicas()
+        {
+            return replicas;
+        }
+
+        public boolean sameGroup(Unit n1, Unit n2)
+        {
+            return false;
+        }
+
+        public Unit getGroup(Unit unit)
+        {
+            // The unit is the group.
+            return unit;
+        }
+
+        public double spreadExpectation()
+        {
+            return 1;
+        }
+    }
+
+    static abstract class GroupReplicationStrategy implements 
TestReplicationStrategy
+    {
+        final int replicas;
+        final Map<Unit, Integer> groupMap;
+
+        public GroupReplicationStrategy(int replicas)
+        {
+            this.replicas = replicas;
+            this.groupMap = Maps.newHashMap();
+        }
+
+        public List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> 
sortedTokens)
+        {
+            List<Unit> endpoints = new ArrayList<Unit>(replicas);
+            BitSet usedGroups = new BitSet();
+
+            if (sortedTokens.isEmpty())
+                return endpoints;
+
+            token = sortedTokens.ceilingKey(token);
+            if (token == null)
+                token = sortedTokens.firstKey();
+            Iterator<Unit> iter = Iterables.concat(sortedTokens.tailMap(token, 
true).values(), sortedTokens.values()).iterator();
+            while (endpoints.size() < replicas)
+            {
+                // For simlicity assuming list can't be exhausted before 
finding all replicas.
+                Unit ep = iter.next();
+                int group = groupMap.get(ep);
+                if (!usedGroups.get(group))
+                {
+                    endpoints.add(ep);
+                    usedGroups.set(group);
+                }
+            }
+            return endpoints;
+        }
+
+        public Token lastReplicaToken(Token token, NavigableMap<Token, Unit> 
sortedTokens)
+        {
+            BitSet usedGroups = new BitSet();
+            int groupsFound = 0;
+
+            token = sortedTokens.ceilingKey(token);
+            if (token == null)
+                token = sortedTokens.firstKey();
+            for (Map.Entry<Token, Unit> en :
+            Iterables.concat(sortedTokens.tailMap(token, true).entrySet(),
+                             sortedTokens.entrySet()))
+            {
+                Unit ep = en.getValue();
+                int group = groupMap.get(ep);
+                if (!usedGroups.get(group))
+                {
+                    usedGroups.set(group);
+                    if (++groupsFound >= replicas)
+                        return en.getKey();
+                }
+            }
+            return token;
+        }
+
+        public Token replicationStart(Token token, Unit unit, 
NavigableMap<Token, Unit> sortedTokens)
+        {
+            // replicated ownership
+            int unitGroup = groupMap.get(unit);   // unit must be already added
+            BitSet seenGroups = new BitSet();
+            int groupsFound = 0;
+
+            for (Map.Entry<Token, Unit> en : Iterables.concat(
+                                                             
sortedTokens.headMap(token, false).descendingMap().entrySet(),
+                                                             
sortedTokens.descendingMap().entrySet()))
+            {
+                Unit n = en.getValue();
+                int ngroup = groupMap.get(n);
+                // Same group as investigated unit is a break; anything that 
could replicate in it replicates there.
+                if (ngroup == unitGroup)
+                    break;
+
+                if (!seenGroups.get(ngroup))
+                {
+                    if (++groupsFound == replicas)
+                        break;
+                    seenGroups.set(ngroup);
+                }
+                token = en.getKey();
+            }
+            return token;
+        }
+
+        public String toString()
+        {
+            Map<Integer, Integer> idToSize = instanceToCount(groupMap);
+            Map<Integer, Integer> sizeToCount = Maps.newTreeMap();
+            sizeToCount.putAll(instanceToCount(idToSize));
+            return String.format("%s strategy, %d replicas, group size to 
count %s", getClass().getSimpleName(), replicas, sizeToCount);
+        }
+
+        @Override
+        public int replicas()
+        {
+            return replicas;
+        }
+
+        public boolean sameGroup(Unit n1, Unit n2)
+        {
+            return groupMap.get(n1).equals(groupMap.get(n2));
+        }
+
+        public void removeUnit(Unit n)
+        {
+            groupMap.remove(n);
+        }
+
+        public Integer getGroup(Unit unit)
+        {
+            return groupMap.get(unit);
+        }
+
+        public double spreadExpectation()
+        {
+            return 1.5;   // Even balanced racks get disbalanced when they 
lose nodes.
+        }
+    }
+
+    private static <T> Map<T, Integer> instanceToCount(Map<?, T> map)
+    {
+        Map<T, Integer> idToCount = Maps.newHashMap();
+        for (Map.Entry<?, T> en : map.entrySet())
+        {
+            Integer old = idToCount.get(en.getValue());
+            idToCount.put(en.getValue(), old != null ? old + 1 : 1);
+        }
+        return idToCount;
+    }
+
+    /**
+     * Group strategy spreading units into a fixed number of groups.
+     */
+    static class FixedGroupCountReplicationStrategy extends 
GroupReplicationStrategy
+    {
+        int groupId;
+        int groupCount;
+
+        public FixedGroupCountReplicationStrategy(int replicas, int groupCount)
+        {
+            super(replicas);
+            assert groupCount >= replicas;
+            groupId = 0;
+            this.groupCount = groupCount;
+        }
+
+        public void addUnit(Unit n)
+        {
+            groupMap.put(n, groupId++ % groupCount);
+        }
+    }
+
+    /**
+     * Group strategy with a fixed number of units per group.
+     */
+    static class BalancedGroupReplicationStrategy extends 
GroupReplicationStrategy
+    {
+        int groupId;
+        int groupSize;
+
+        public BalancedGroupReplicationStrategy(int replicas, int groupSize)
+        {
+            super(replicas);
+            groupId = 0;
+            this.groupSize = groupSize;
+        }
+
+        public void addUnit(Unit n)
+        {
+            groupMap.put(n, groupId++ / groupSize);
+        }
+    }
+
+    static class UnbalancedGroupReplicationStrategy extends 
GroupReplicationStrategy
+    {
+        int groupId;
+        int nextSize;
+        int num;
+        int minGroupSize;
+        int maxGroupSize;
+        Random rand;
+
+        public UnbalancedGroupReplicationStrategy(int replicas, int 
minGroupSize, int maxGroupSize, Random rand)
+        {
+            super(replicas);
+            groupId = -1;
+            nextSize = 0;
+            num = 0;
+            this.maxGroupSize = maxGroupSize;
+            this.minGroupSize = minGroupSize;
+            this.rand = rand;
+        }
+
+        public void addUnit(Unit n)
+        {
+            if (++num > nextSize)
+            {
+                nextSize = minGroupSize + rand.nextInt(maxGroupSize - 
minGroupSize + 1);
+                ++groupId;
+                num = 0;
+            }
+            groupMap.put(n, groupId);
+        }
+
+        public double spreadExpectation()
+        {
+            return 2;
+        }
+    }
+
+    static Map<Unit, Double> 
evaluateReplicatedOwnership(ReplicationAwareTokenAllocator<Unit> t)
+    {
+        Map<Unit, Double> ownership = Maps.newHashMap();
+        Iterator<Token> it = t.sortedTokens.keySet().iterator();
+        if (!it.hasNext())
+            return ownership;
+
+        Token current = it.next();
+        while (it.hasNext())
+        {
+            Token next = it.next();
+            addOwnership(t, current, next, ownership);
+            current = next;
+        }
+        addOwnership(t, current, t.sortedTokens.firstKey(), ownership);
+
+        return ownership;
+    }
+
+    private static void addOwnership(ReplicationAwareTokenAllocator<Unit> t, 
Token current, Token next, Map<Unit, Double> ownership)
+    {
+        TestReplicationStrategy ts = (TestReplicationStrategy) t.strategy;
+        double size = current.size(next);
+        Token representative = t.partitioner.midpoint(current, next);
+        for (Unit n : ts.getReplicas(representative, t.sortedTokens))
+        {
+            Double v = ownership.get(n);
+            ownership.put(n, v != null ? v + size : size);
+        }
+    }
+
+    private static double replicatedTokenOwnership(Token token, 
NavigableMap<Token, Unit> sortedTokens, ReplicationStrategy<Unit> strategy)
+    {
+        TestReplicationStrategy ts = (TestReplicationStrategy) strategy;
+        Token next = sortedTokens.higherKey(token);
+        if (next == null)
+            next = sortedTokens.firstKey();
+        return ts.replicationStart(token, sortedTokens.get(token), 
sortedTokens).size(next);
+    }
+
+    static interface TokenCount
+    {
+        int tokenCount(int perUnitCount, Random rand);
+
+        double spreadExpectation();
+    }
+
+    static TokenCount fixedTokenCount = new TokenCount()
+    {
+        public int tokenCount(int perUnitCount, Random rand)
+        {
+            return perUnitCount;
+        }
+
+        public double spreadExpectation()
+        {
+            return 4;  // High tolerance to avoid flakiness.
+        }
+    };
+
+    static TokenCount varyingTokenCount = new TokenCount()
+    {
+        public int tokenCount(int perUnitCount, Random rand)
+        {
+            if (perUnitCount == 1) return 1;
+            // 25 to 175%
+            return rand.nextInt(perUnitCount * 3 / 2) + (perUnitCount + 3) / 4;
+        }
+
+        public double spreadExpectation()
+        {
+            return 8;  // High tolerance to avoid flakiness.
+        }
+    };
+
+    Random seededRand = new Random(2);
+
+    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;
+        for (int i = 0; i < unitCount; i++)
+        {
+            Unit unit = new Unit();
+            rs.addUnit(unit);
+            int tokens = tc.tokenCount(perUnitCount, rand);
+            for (int j = 0; j < tokens; j++)
+            {
+                map.put(partitioner.getRandomToken(rand), unit);
+            }
+        }
+    }
+
+    protected void testExistingCluster(IPartitioner partitioner, int 
maxVNodeCount)
+    {
+        for (int rf = 1; rf <= 5; ++rf)
+        {
+            for (int perUnitCount = 1; perUnitCount <= maxVNodeCount; 
perUnitCount *= 4)
+            {
+                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), partitioner);
+                    testExistingCluster(perUnitCount, varyingTokenCount,
+                                        new 
UnbalancedGroupReplicationStrategy(rf, groupSize / 2, groupSize * 2, 
seededRand),
+                                        partitioner);
+                }
+                testExistingCluster(perUnitCount, fixedTokenCount,
+                                    new FixedGroupCountReplicationStrategy(rf, 
rf * 2), partitioner);
+            }
+        }
+    }
+
+    private 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, 
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, 
partitioner);
+        System.out.println();
+    }
+
+    protected void testNewCluster(IPartitioner partitioner, int maxVNodeCount)
+    {
+        // 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)
+        // ideal distribution of tokens for some number of nodes, which in 
turn will inevitably cause it to go into a
+        // bad (unacceptable to the test criteria) distribution after adding 
one more node.
+
+        // This should happen very rarely, unless something is broken in the 
token allocation code.
+
+        for (int rf = 2; rf <= 5; ++rf)
+        {
+            for (int perUnitCount = 1; perUnitCount <= maxVNodeCount; 
perUnitCount *= 4)
+            {
+                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), partitioner);
+                    testNewCluster(perUnitCount, varyingTokenCount,
+                                   new UnbalancedGroupReplicationStrategy(rf, 
groupSize / 2, groupSize * 2, seededRand),
+                                   partitioner);
+                }
+                testNewCluster(perUnitCount, fixedTokenCount,
+                               new FixedGroupCountReplicationStrategy(rf, rf * 
2), partitioner);
+            }
+        }
+    }
+
+    private 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;
+        NavigableMap<Token, Unit> tokenMap = Maps.newTreeMap();
+
+        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, 
partitioner);
+        System.out.println();
+    }
+
+    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);
+        for (int i = 0; i < howMany; ++i)
+        {
+            Unit u = t.unitFor(partitioner.getRandomToken(seededRand));
+            t.removeUnit(u);
+            ((TestReplicationStrategy) t.strategy).removeUnit(u);
+        }
+        // Grow half without verifying.
+        grow(t, (t.unitCount() + fullCount * 3) / 4, tc, perUnitCount, false);
+        // Metrics should be back to normal by now. Check that they remain so.
+        grow(t, fullCount, tc, perUnitCount, true);
+    }
+
+    static class Summary
+    {
+        double min = 1;
+        double max = 1;
+        double stddev = 0;
+
+        void update(SummaryStatistics stat)
+        {
+            min = Math.min(min, stat.getMin());
+            max = Math.max(max, stat.getMax());
+            stddev = Math.max(stddev, stat.getStandardDeviation());
+        }
+
+        public String toString()
+        {
+            return String.format("max %.2f min %.2f stddev %.4f", max, min, 
stddev);
+        }
+    }
+
+    public void grow(ReplicationAwareTokenAllocator<Unit> t, int 
targetClusterSize, TokenCount tc, int perUnitCount, boolean verifyMetrics)
+    {
+        int size = t.unitCount();
+        Summary su = new Summary();
+        Summary st = new Summary();
+        Random rand = new Random(targetClusterSize + perUnitCount);
+        TestReplicationStrategy strategy = (TestReplicationStrategy) 
t.strategy;
+        if (size < targetClusterSize)
+        {
+            System.out.format("Adding %d unit(s) using %s...", 
targetClusterSize - size, t.toString());
+            long time = System.currentTimeMillis();
+            while (size < targetClusterSize)
+            {
+                int tokens = tc.tokenCount(perUnitCount, rand);
+                Unit unit = new Unit();
+                strategy.addUnit(unit);
+                t.addUnit(unit, tokens);
+                ++size;
+                if (verifyMetrics)
+                    updateSummary(t, su, st, false);
+            }
+            System.out.format(" Done in %.3fs\n", (System.currentTimeMillis() 
- time) / 1000.0);
+            if (verifyMetrics)
+            {
+                updateSummary(t, su, st, true);
+                double maxExpected = 1.0 + tc.spreadExpectation() * 
strategy.spreadExpectation() / (perUnitCount * t.replicas);
+                if (su.max > maxExpected)
+                {
+                    Assert.fail(String.format("Expected max unit size below 
%.4f, was %.4f", maxExpected, su.max));
+                }
+                // We can't verify lower side range as small loads can't 
always be fixed.
+            }
+        }
+    }
+
+
+    private void updateSummary(ReplicationAwareTokenAllocator<Unit> t, Summary 
su, Summary st, boolean print)
+    {
+        int size = t.sortedTokens.size();
+        double inverseAverage = 1.0 * size / t.strategy.replicas();
+
+        Map<Unit, Double> ownership = evaluateReplicatedOwnership(t);
+        SummaryStatistics unitStat = new SummaryStatistics();
+        for (Map.Entry<Unit, Double> en : ownership.entrySet())
+            unitStat.addValue(en.getValue() * inverseAverage / 
t.unitToTokens.get(en.getKey()).size());
+        su.update(unitStat);
+
+        SummaryStatistics tokenStat = new SummaryStatistics();
+        for (Token tok : t.sortedTokens.keySet())
+            tokenStat.addValue(replicatedTokenOwnership(tok, t.sortedTokens, 
t.strategy) * inverseAverage);
+        st.update(tokenStat);
+
+        if (print)
+        {
+            System.out.format("Size %d(%d)   \tunit %s  token %s   %s\n",
+                              t.unitCount(), size,
+                              mms(unitStat),
+                              mms(tokenStat),
+                              t.strategy);
+            System.out.format("Worst intermediate unit\t%s  token %s\n", su, 
st);
+        }
+    }
+
+
+    private static String mms(SummaryStatistics s)
+    {
+        return String.format("max %.2f min %.2f stddev %.4f", s.getMax(), 
s.getMin(), s.getStandardDeviation());
+    }
+
+
+    int nextUnitId = 0;
+
+    final class Unit implements Comparable<Unit>
+    {
+        int unitId = nextUnitId++;
+
+        public String toString()
+        {
+            return Integer.toString(unitId);
+        }
+
+        @Override
+        public int compareTo(Unit o)
+        {
+            return Integer.compare(unitId, o.unitId);
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c6ec31ba/test/long/org/apache/cassandra/dht/tokenallocator/Murmur3ReplicationAwareTokenAllocatorTest.java
----------------------------------------------------------------------
diff --git 
a/test/long/org/apache/cassandra/dht/tokenallocator/Murmur3ReplicationAwareTokenAllocatorTest.java
 
b/test/long/org/apache/cassandra/dht/tokenallocator/Murmur3ReplicationAwareTokenAllocatorTest.java
new file mode 100644
index 0000000..e28ecfa
--- /dev/null
+++ 
b/test/long/org/apache/cassandra/dht/tokenallocator/Murmur3ReplicationAwareTokenAllocatorTest.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.dht.tokenallocator;
+
+import org.junit.Test;
+
+import org.apache.cassandra.Util;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+
+public class Murmur3ReplicationAwareTokenAllocatorTest extends 
AbstractReplicationAwareTokenAllocatorTest
+{
+    private static final int MAX_VNODE_COUNT = 64;
+
+    @Test
+    public void testExistingCluster()
+    {
+        super.testExistingCluster(new Murmur3Partitioner(), MAX_VNODE_COUNT);
+    }
+
+    @Test
+    public void testNewCluster()
+    {
+        Util.flakyTest(this::flakyTestNewCluster,
+                       2,
+                       "It tends to fail sometimes due to the random selection 
of the tokens in the first few nodes.");
+    }
+
+    private void flakyTestNewCluster()
+    {
+        testNewCluster(new Murmur3Partitioner(), MAX_VNODE_COUNT);
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c6ec31ba/test/long/org/apache/cassandra/dht/tokenallocator/RandomReplicationAwareTokenAllocatorTest.java
----------------------------------------------------------------------
diff --git 
a/test/long/org/apache/cassandra/dht/tokenallocator/RandomReplicationAwareTokenAllocatorTest.java
 
b/test/long/org/apache/cassandra/dht/tokenallocator/RandomReplicationAwareTokenAllocatorTest.java
new file mode 100644
index 0000000..bd94442
--- /dev/null
+++ 
b/test/long/org/apache/cassandra/dht/tokenallocator/RandomReplicationAwareTokenAllocatorTest.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.dht.tokenallocator;
+
+import org.junit.Test;
+
+import org.apache.cassandra.Util;
+import org.apache.cassandra.dht.RandomPartitioner;
+
+public class RandomReplicationAwareTokenAllocatorTest extends 
AbstractReplicationAwareTokenAllocatorTest
+{
+    /** The maximum number of vnodes to use in the tests.
+     *  For RandomPartitioner we use a smaller number because
+     *  the tests take much longer and would otherwise timeout,
+     *  see CASSANDRA-12784.
+     * */
+    private static final int MAX_VNODE_COUNT = 16;
+
+    @Test
+    public void testExistingCluster()
+    {
+        testExistingCluster(new RandomPartitioner(), MAX_VNODE_COUNT);
+    }
+
+    @Test
+    public void testNewClusterr()
+    {
+        Util.flakyTest(this::flakyTestNewCluster,
+                       3,
+                       "It tends to fail sometimes due to the random selection 
of the tokens in the first few nodes.");
+    }
+
+    private void flakyTestNewCluster()
+    {
+        testNewCluster(new RandomPartitioner(), MAX_VNODE_COUNT);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c6ec31ba/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
deleted file mode 100644
index 482e2ac..0000000
--- 
a/test/long/org/apache/cassandra/dht/tokenallocator/ReplicationAwareTokenAllocatorTest.java
+++ /dev/null
@@ -1,755 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.cassandra.dht.tokenallocator;
-
-import java.util.*;
-
-import junit.framework.Assert;
-
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-
-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
-{
-    private static final int MAX_VNODE_COUNT = 64;
-
-    private static final int TARGET_CLUSTER_SIZE = 250;
-
-    interface TestReplicationStrategy extends ReplicationStrategy<Unit>
-    {
-        void addUnit(Unit n);
-
-        void removeUnit(Unit n);
-
-        /**
-         * Returns a list of all replica units for given token.
-         */
-        List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> 
sortedTokens);
-
-        /**
-         * Returns the start of the token span that is replicated in this 
token.
-         * Note: Though this is not trivial to see, the replicated span is 
always contiguous. A token in the same
-         * group acts as a barrier; if one is not found the token replicates 
everything up to the replica'th distinct
-         * group seen in front of it.
-         */
-        Token replicationStart(Token token, Unit unit, NavigableMap<Token, 
Unit> sortedTokens);
-
-        /**
-         * Multiplier for the acceptable disbalance in the cluster. With some 
strategies it is harder to achieve good
-         * results.
-         */
-        public double spreadExpectation();
-    }
-
-    static class NoReplicationStrategy implements TestReplicationStrategy
-    {
-        public List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> 
sortedTokens)
-        {
-            return 
Collections.singletonList(sortedTokens.ceilingEntry(token).getValue());
-        }
-
-        public Token replicationStart(Token token, Unit unit, 
NavigableMap<Token, Unit> sortedTokens)
-        {
-            return sortedTokens.lowerKey(token);
-        }
-
-        public String toString()
-        {
-            return "No replication";
-        }
-
-        public void addUnit(Unit n)
-        {
-        }
-
-        public void removeUnit(Unit n)
-        {
-        }
-
-        public int replicas()
-        {
-            return 1;
-        }
-
-        public boolean sameGroup(Unit n1, Unit n2)
-        {
-            return false;
-        }
-
-        public Object getGroup(Unit unit)
-        {
-            return unit;
-        }
-
-        public double spreadExpectation()
-        {
-            return 1;
-        }
-    }
-
-    static class SimpleReplicationStrategy implements TestReplicationStrategy
-    {
-        int replicas;
-
-        public SimpleReplicationStrategy(int replicas)
-        {
-            super();
-            this.replicas = replicas;
-        }
-
-        public List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> 
sortedTokens)
-        {
-            List<Unit> endpoints = new ArrayList<Unit>(replicas);
-
-            token = sortedTokens.ceilingKey(token);
-            if (token == null)
-                token = sortedTokens.firstKey();
-            Iterator<Unit> iter = Iterables.concat(sortedTokens.tailMap(token, 
true).values(), sortedTokens.values()).iterator();
-            while (endpoints.size() < replicas)
-            {
-                if (!iter.hasNext())
-                    return endpoints;
-                Unit ep = iter.next();
-                if (!endpoints.contains(ep))
-                    endpoints.add(ep);
-            }
-            return endpoints;
-        }
-
-        public Token replicationStart(Token token, Unit unit, 
NavigableMap<Token, Unit> sortedTokens)
-        {
-            Set<Unit> seenUnits = Sets.newHashSet();
-            int unitsFound = 0;
-
-            for (Map.Entry<Token, Unit> en : Iterables.concat(
-                                                             
sortedTokens.headMap(token, false).descendingMap().entrySet(),
-                                                             
sortedTokens.descendingMap().entrySet()))
-            {
-                Unit n = en.getValue();
-                // Same group as investigated unit is a break; anything that 
could replicate in it replicates there.
-                if (n == unit)
-                    break;
-
-                if (seenUnits.add(n))
-                {
-                    if (++unitsFound == replicas)
-                        break;
-                }
-                token = en.getKey();
-            }
-            return token;
-        }
-
-        public void addUnit(Unit n)
-        {
-        }
-
-        public void removeUnit(Unit n)
-        {
-        }
-
-        public String toString()
-        {
-            return String.format("Simple %d replicas", replicas);
-        }
-
-        public int replicas()
-        {
-            return replicas;
-        }
-
-        public boolean sameGroup(Unit n1, Unit n2)
-        {
-            return false;
-        }
-
-        public Unit getGroup(Unit unit)
-        {
-            // The unit is the group.
-            return unit;
-        }
-
-        public double spreadExpectation()
-        {
-            return 1;
-        }
-    }
-
-    static abstract class GroupReplicationStrategy implements 
TestReplicationStrategy
-    {
-        final int replicas;
-        final Map<Unit, Integer> groupMap;
-
-        public GroupReplicationStrategy(int replicas)
-        {
-            this.replicas = replicas;
-            this.groupMap = Maps.newHashMap();
-        }
-
-        public List<Unit> getReplicas(Token token, NavigableMap<Token, Unit> 
sortedTokens)
-        {
-            List<Unit> endpoints = new ArrayList<Unit>(replicas);
-            BitSet usedGroups = new BitSet();
-
-            if (sortedTokens.isEmpty())
-                return endpoints;
-
-            token = sortedTokens.ceilingKey(token);
-            if (token == null)
-                token = sortedTokens.firstKey();
-            Iterator<Unit> iter = Iterables.concat(sortedTokens.tailMap(token, 
true).values(), sortedTokens.values()).iterator();
-            while (endpoints.size() < replicas)
-            {
-                // For simlicity assuming list can't be exhausted before 
finding all replicas.
-                Unit ep = iter.next();
-                int group = groupMap.get(ep);
-                if (!usedGroups.get(group))
-                {
-                    endpoints.add(ep);
-                    usedGroups.set(group);
-                }
-            }
-            return endpoints;
-        }
-
-        public Token lastReplicaToken(Token token, NavigableMap<Token, Unit> 
sortedTokens)
-        {
-            BitSet usedGroups = new BitSet();
-            int groupsFound = 0;
-
-            token = sortedTokens.ceilingKey(token);
-            if (token == null)
-                token = sortedTokens.firstKey();
-            for (Map.Entry<Token, Unit> en :
-            Iterables.concat(sortedTokens.tailMap(token, true).entrySet(),
-                             sortedTokens.entrySet()))
-            {
-                Unit ep = en.getValue();
-                int group = groupMap.get(ep);
-                if (!usedGroups.get(group))
-                {
-                    usedGroups.set(group);
-                    if (++groupsFound >= replicas)
-                        return en.getKey();
-                }
-            }
-            return token;
-        }
-
-        public Token replicationStart(Token token, Unit unit, 
NavigableMap<Token, Unit> sortedTokens)
-        {
-            // replicated ownership
-            int unitGroup = groupMap.get(unit);   // unit must be already added
-            BitSet seenGroups = new BitSet();
-            int groupsFound = 0;
-
-            for (Map.Entry<Token, Unit> en : Iterables.concat(
-                                                             
sortedTokens.headMap(token, false).descendingMap().entrySet(),
-                                                             
sortedTokens.descendingMap().entrySet()))
-            {
-                Unit n = en.getValue();
-                int ngroup = groupMap.get(n);
-                // Same group as investigated unit is a break; anything that 
could replicate in it replicates there.
-                if (ngroup == unitGroup)
-                    break;
-
-                if (!seenGroups.get(ngroup))
-                {
-                    if (++groupsFound == replicas)
-                        break;
-                    seenGroups.set(ngroup);
-                }
-                token = en.getKey();
-            }
-            return token;
-        }
-
-        public String toString()
-        {
-            Map<Integer, Integer> idToSize = instanceToCount(groupMap);
-            Map<Integer, Integer> sizeToCount = Maps.newTreeMap();
-            sizeToCount.putAll(instanceToCount(idToSize));
-            return String.format("%s strategy, %d replicas, group size to 
count %s", getClass().getSimpleName(), replicas, sizeToCount);
-        }
-
-        @Override
-        public int replicas()
-        {
-            return replicas;
-        }
-
-        public boolean sameGroup(Unit n1, Unit n2)
-        {
-            return groupMap.get(n1).equals(groupMap.get(n2));
-        }
-
-        public void removeUnit(Unit n)
-        {
-            groupMap.remove(n);
-        }
-
-        public Integer getGroup(Unit unit)
-        {
-            return groupMap.get(unit);
-        }
-
-        public double spreadExpectation()
-        {
-            return 1.5;   // Even balanced racks get disbalanced when they 
lose nodes.
-        }
-    }
-
-    private static <T> Map<T, Integer> instanceToCount(Map<?, T> map)
-    {
-        Map<T, Integer> idToCount = Maps.newHashMap();
-        for (Map.Entry<?, T> en : map.entrySet())
-        {
-            Integer old = idToCount.get(en.getValue());
-            idToCount.put(en.getValue(), old != null ? old + 1 : 1);
-        }
-        return idToCount;
-    }
-
-    /**
-     * Group strategy spreading units into a fixed number of groups.
-     */
-    static class FixedGroupCountReplicationStrategy extends 
GroupReplicationStrategy
-    {
-        int groupId;
-        int groupCount;
-
-        public FixedGroupCountReplicationStrategy(int replicas, int groupCount)
-        {
-            super(replicas);
-            assert groupCount >= replicas;
-            groupId = 0;
-            this.groupCount = groupCount;
-        }
-
-        public void addUnit(Unit n)
-        {
-            groupMap.put(n, groupId++ % groupCount);
-        }
-    }
-
-    /**
-     * Group strategy with a fixed number of units per group.
-     */
-    static class BalancedGroupReplicationStrategy extends 
GroupReplicationStrategy
-    {
-        int groupId;
-        int groupSize;
-
-        public BalancedGroupReplicationStrategy(int replicas, int groupSize)
-        {
-            super(replicas);
-            groupId = 0;
-            this.groupSize = groupSize;
-        }
-
-        public void addUnit(Unit n)
-        {
-            groupMap.put(n, groupId++ / groupSize);
-        }
-    }
-
-    static class UnbalancedGroupReplicationStrategy extends 
GroupReplicationStrategy
-    {
-        int groupId;
-        int nextSize;
-        int num;
-        int minGroupSize;
-        int maxGroupSize;
-        Random rand;
-
-        public UnbalancedGroupReplicationStrategy(int replicas, int 
minGroupSize, int maxGroupSize, Random rand)
-        {
-            super(replicas);
-            groupId = -1;
-            nextSize = 0;
-            num = 0;
-            this.maxGroupSize = maxGroupSize;
-            this.minGroupSize = minGroupSize;
-            this.rand = rand;
-        }
-
-        public void addUnit(Unit n)
-        {
-            if (++num > nextSize)
-            {
-                nextSize = minGroupSize + rand.nextInt(maxGroupSize - 
minGroupSize + 1);
-                ++groupId;
-                num = 0;
-            }
-            groupMap.put(n, groupId);
-        }
-
-        public double spreadExpectation()
-        {
-            return 2;
-        }
-    }
-
-    static Map<Unit, Double> 
evaluateReplicatedOwnership(ReplicationAwareTokenAllocator<Unit> t)
-    {
-        Map<Unit, Double> ownership = Maps.newHashMap();
-        Iterator<Token> it = t.sortedTokens.keySet().iterator();
-        if (!it.hasNext())
-            return ownership;
-
-        Token current = it.next();
-        while (it.hasNext())
-        {
-            Token next = it.next();
-            addOwnership(t, current, next, ownership);
-            current = next;
-        }
-        addOwnership(t, current, t.sortedTokens.firstKey(), ownership);
-
-        return ownership;
-    }
-
-    private static void addOwnership(ReplicationAwareTokenAllocator<Unit> t, 
Token current, Token next, Map<Unit, Double> ownership)
-    {
-        TestReplicationStrategy ts = (TestReplicationStrategy) t.strategy;
-        double size = current.size(next);
-        Token representative = t.partitioner.midpoint(current, next);
-        for (Unit n : ts.getReplicas(representative, t.sortedTokens))
-        {
-            Double v = ownership.get(n);
-            ownership.put(n, v != null ? v + size : size);
-        }
-    }
-
-    private static double replicatedTokenOwnership(Token token, 
NavigableMap<Token, Unit> sortedTokens, ReplicationStrategy<Unit> strategy)
-    {
-        TestReplicationStrategy ts = (TestReplicationStrategy) strategy;
-        Token next = sortedTokens.higherKey(token);
-        if (next == null)
-            next = sortedTokens.firstKey();
-        return ts.replicationStart(token, sortedTokens.get(token), 
sortedTokens).size(next);
-    }
-
-    static interface TokenCount
-    {
-        int tokenCount(int perUnitCount, Random rand);
-
-        double spreadExpectation();
-    }
-
-    static TokenCount fixedTokenCount = new TokenCount()
-    {
-        public int tokenCount(int perUnitCount, Random rand)
-        {
-            return perUnitCount;
-        }
-
-        public double spreadExpectation()
-        {
-            return 4;  // High tolerance to avoid flakiness.
-        }
-    };
-
-    static TokenCount varyingTokenCount = new TokenCount()
-    {
-        public int tokenCount(int perUnitCount, Random rand)
-        {
-            if (perUnitCount == 1) return 1;
-            // 25 to 175%
-            return rand.nextInt(perUnitCount * 3 / 2) + (perUnitCount + 3) / 4;
-        }
-
-        public double spreadExpectation()
-        {
-            return 8;  // High tolerance to avoid flakiness.
-        }
-    };
-
-    Random seededRand = new Random(2);
-
-    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;
-        for (int i = 0; i < unitCount; i++)
-        {
-            Unit unit = new Unit();
-            rs.addUnit(unit);
-            int tokens = tc.tokenCount(perUnitCount, rand);
-            for (int j = 0; j < tokens; j++)
-            {
-                map.put(partitioner.getRandomToken(rand), unit);
-            }
-        }
-    }
-
-    @Test
-    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), 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), partitioner);
-                    testExistingCluster(perUnitCount, varyingTokenCount,
-                                        new 
UnbalancedGroupReplicationStrategy(rf, groupSize / 2, groupSize * 2, 
seededRand),
-                                        partitioner);
-                }
-                testExistingCluster(perUnitCount, fixedTokenCount,
-                                    new FixedGroupCountReplicationStrategy(rf, 
rf * 2), partitioner);
-            }
-        }
-    }
-
-    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, 
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, 
partitioner);
-        System.out.println();
-    }
-
-    @Test
-    public void testNewClusterWithRandomPartitioner()
-    {
-        Util.flakyTest(this::flakyTestNewClusterWithRandomPartitioner,
-                       5,
-                       "It tends to fail sometimes due to the random selection 
of the tokens in the first few nodes.");
-    }
-
-    @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)
-        // ideal distribution of tokens for some number of nodes, which in 
turn will inevitably cause it to go into a
-        // bad (unacceptable to the test criteria) distribution after adding 
one more node.
-
-        // This should happen very rarely, unless something is broken in the 
token allocation code.
-
-        for (int rf = 2; rf <= 5; ++rf)
-        {
-            for (int perUnitCount = 1; perUnitCount <= MAX_VNODE_COUNT; 
perUnitCount *= 4)
-            {
-                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), partitioner);
-                    testNewCluster(perUnitCount, varyingTokenCount,
-                                   new UnbalancedGroupReplicationStrategy(rf, 
groupSize / 2, groupSize * 2, seededRand),
-                                   partitioner);
-                }
-                testNewCluster(perUnitCount, fixedTokenCount,
-                               new FixedGroupCountReplicationStrategy(rf, rf * 
2), partitioner);
-            }
-        }
-    }
-
-    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;
-        NavigableMap<Token, Unit> tokenMap = Maps.newTreeMap();
-
-        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, 
partitioner);
-        System.out.println();
-    }
-
-    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);
-        for (int i = 0; i < howMany; ++i)
-        {
-            Unit u = t.unitFor(partitioner.getRandomToken(seededRand));
-            t.removeUnit(u);
-            ((TestReplicationStrategy) t.strategy).removeUnit(u);
-        }
-        // Grow half without verifying.
-        grow(t, (t.unitCount() + fullCount * 3) / 4, tc, perUnitCount, false);
-        // Metrics should be back to normal by now. Check that they remain so.
-        grow(t, fullCount, tc, perUnitCount, true);
-    }
-
-    static class Summary
-    {
-        double min = 1;
-        double max = 1;
-        double stddev = 0;
-
-        void update(SummaryStatistics stat)
-        {
-            min = Math.min(min, stat.getMin());
-            max = Math.max(max, stat.getMax());
-            stddev = Math.max(stddev, stat.getStandardDeviation());
-        }
-
-        public String toString()
-        {
-            return String.format("max %.2f min %.2f stddev %.4f", max, min, 
stddev);
-        }
-    }
-
-    public void grow(ReplicationAwareTokenAllocator<Unit> t, int 
targetClusterSize, TokenCount tc, int perUnitCount, boolean verifyMetrics)
-    {
-        int size = t.unitCount();
-        Summary su = new Summary();
-        Summary st = new Summary();
-        Random rand = new Random(targetClusterSize + perUnitCount);
-        TestReplicationStrategy strategy = (TestReplicationStrategy) 
t.strategy;
-        if (size < targetClusterSize)
-        {
-            System.out.format("Adding %d unit(s) using %s...", 
targetClusterSize - size, t.toString());
-            long time = System.currentTimeMillis();
-            while (size < targetClusterSize)
-            {
-                int tokens = tc.tokenCount(perUnitCount, rand);
-                Unit unit = new Unit();
-                strategy.addUnit(unit);
-                t.addUnit(unit, tokens);
-                ++size;
-                if (verifyMetrics)
-                    updateSummary(t, su, st, false);
-            }
-            System.out.format(" Done in %.3fs\n", (System.currentTimeMillis() 
- time) / 1000.0);
-            if (verifyMetrics)
-            {
-                updateSummary(t, su, st, true);
-                double maxExpected = 1.0 + tc.spreadExpectation() * 
strategy.spreadExpectation() / (perUnitCount * t.replicas);
-                if (su.max > maxExpected)
-                {
-                    Assert.fail(String.format("Expected max unit size below 
%.4f, was %.4f", maxExpected, su.max));
-                }
-                // We can't verify lower side range as small loads can't 
always be fixed.
-            }
-        }
-    }
-
-
-    private void updateSummary(ReplicationAwareTokenAllocator<Unit> t, Summary 
su, Summary st, boolean print)
-    {
-        int size = t.sortedTokens.size();
-        double inverseAverage = 1.0 * size / t.strategy.replicas();
-
-        Map<Unit, Double> ownership = evaluateReplicatedOwnership(t);
-        SummaryStatistics unitStat = new SummaryStatistics();
-        for (Map.Entry<Unit, Double> en : ownership.entrySet())
-            unitStat.addValue(en.getValue() * inverseAverage / 
t.unitToTokens.get(en.getKey()).size());
-        su.update(unitStat);
-
-        SummaryStatistics tokenStat = new SummaryStatistics();
-        for (Token tok : t.sortedTokens.keySet())
-            tokenStat.addValue(replicatedTokenOwnership(tok, t.sortedTokens, 
t.strategy) * inverseAverage);
-        st.update(tokenStat);
-
-        if (print)
-        {
-            System.out.format("Size %d(%d)   \tunit %s  token %s   %s\n",
-                              t.unitCount(), size,
-                              mms(unitStat),
-                              mms(tokenStat),
-                              t.strategy);
-            System.out.format("Worst intermediate unit\t%s  token %s\n", su, 
st);
-        }
-    }
-
-
-    private static String mms(SummaryStatistics s)
-    {
-        return String.format("max %.2f min %.2f stddev %.4f", s.getMax(), 
s.getMin(), s.getStandardDeviation());
-    }
-
-
-    int nextUnitId = 0;
-
-    final class Unit implements Comparable<Unit>
-    {
-        int unitId = nextUnitId++;
-
-        public String toString()
-        {
-            return Integer.toString(unitId);
-        }
-
-        @Override
-        public int compareTo(Unit o)
-        {
-            return Integer.compare(unitId, o.unitId);
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/cassandra/blob/c6ec31ba/test/unit/org/apache/cassandra/Util.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/Util.java 
b/test/unit/org/apache/cassandra/Util.java
index 2f083c9..f93ce5c 100644
--- a/test/unit/org/apache/cassandra/Util.java
+++ b/test/unit/org/apache/cassandra/Util.java
@@ -36,6 +36,9 @@ import com.google.common.base.Preconditions;
 import com.google.common.collect.Iterators;
 import org.apache.commons.lang3.StringUtils;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ColumnDefinition;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -67,7 +70,9 @@ import static org.junit.Assert.assertTrue;
 
 public class Util
 {
-    private static List<UUID> hostIdPool = new ArrayList<UUID>();
+    private static final Logger logger = LoggerFactory.getLogger(Util.class);
+
+    private static List<UUID> hostIdPool = new ArrayList<>();
 
     public static IPartitioner testPartitioner()
     {
@@ -601,10 +606,9 @@ public class Util
         AssertionError e = runCatchingAssertionError(test);
         if (e == null)
             return;     // success
-        System.err.format("Test failed. %s%n"
-                        + "Re-running %d times to verify it isn't failing more 
often than it should.%n"
-                        + "Failure was: %s%n", message, rerunsOnFailure, e);
-        e.printStackTrace();
+
+        logger.info("Test failed. {}", message, e);
+        logger.info("Re-running {} times to verify it isn't failing more often 
than it should.", rerunsOnFailure);
 
         int rerunsFailed = 0;
         for (int i = 0; i < rerunsOnFailure; ++i)
@@ -614,15 +618,17 @@ public class Util
             {
                 ++rerunsFailed;
                 e.addSuppressed(t);
+
+                logger.debug("Test failed again, total num failures: {}", 
rerunsFailed, t);
             }
         }
         if (rerunsFailed > 0)
         {
-            System.err.format("Test failed in %d of the %d reruns.%n", 
rerunsFailed, rerunsOnFailure);
+            logger.error("Test failed in {} of the {} reruns.", rerunsFailed, 
rerunsOnFailure);
             throw e;
         }
 
-        System.err.println("All reruns succeeded. Failure treated as flake.");
+        logger.info("All reruns succeeded. Failure treated as flake.");
     }
 
     // for use with Optional in tests, can be used as an argument to 
orElseThrow

Reply via email to