You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by st...@apache.org on 2016/10/18 01:14:39 UTC
[2/3] cassandra git commit: Fix timeout in
ReplicationAwareTokenAllocatorTest
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/trunk
Commit: c6ec31bada33b9b803d09a863414ea9cad72752e
Parents: 7a5118c
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Thu Oct 13 16:32:58 2016 +0800
Committer: Stefania Alborghetti <st...@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