natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Handle MIN_TOKEN placement correctly. 

Patch by Alex Petrov, reviewed by Sam Tunnicliffe for CASSANDRA-19262.
This commit is contained in:
Alex Petrov 2024-01-08 17:09:27 +01:00
parent c5a023a204
commit 1cb6d3568b
10 changed files with 269 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/java/org/apache/cassandra/locator/SimpleStrategy.java
View File

@ -17,7 +17,11 @@
*/
package org.apache.cassandra.locator;
import java.util.*;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -36,9 +40,9 @@ import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.ownership.TokenMap;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.PlacementForRange;
import org.apache.cassandra.tcm.ownership.TokenMap;
import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
/**
@ -159,6 +163,7 @@ public class SimpleStrategy extends AbstractReplicationStrategy
return Collections.singleton(REPLICATION_FACTOR);
}
@SuppressWarnings("unused") // used via reflection
protected static void prepareOptions(Map<String, String> options, Map<String, String> previousOptions)
{
// When altering from NTS to SS, previousOptions could have multiple different RFs for different data centers - so we

5
src/java/org/apache/cassandra/tcm/compatibility/TokenRingUtils.java
View File

@ -63,11 +63,6 @@ public class TokenRingUtils
return i;
}
public static Token firstToken(List<Token> ring, Token start)
{
return ring.get(firstTokenIndex(ring, start, false));
}
public static Token getPredecessor(List<Token> ring, Token start)
{
int idx = firstTokenIndex(ring, start, false);

6
src/java/org/apache/cassandra/tcm/sequences/LockedRanges.java
View File

@ -338,6 +338,12 @@ public class LockedRanges implements MetadataValue<LockedRanges>
{
if (thisRange.intersects(otherRange))
return true;
// Since we allow ownership of the MIN_TOKEN, we need to lock both sides of the
// wraparound range in case it transitions from non-wraparound to wraparound and back.
if ((thisRange.left.isMinimum() || thisRange.right.isMinimum()) &&
(otherRange.left.isMinimum() || otherRange.right.isMinimum()))
return true;
}
}
}

152
test/distributed/org/apache/cassandra/distributed/test/log/MetadataChangeSimulationTest.java
View File

@ -45,9 +45,9 @@ import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.harry.sut.TokenPlacementModel;
import org.apache.cassandra.locator.CMSPlacementStrategy;
import org.apache.cassandra.locator.EndpointsForRange;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.CMSPlacementStrategy;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.tcm.AtomicLongBackedProcessor;
@ -65,8 +65,13 @@ import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
import org.apache.cassandra.tcm.transformations.Register;
import org.apache.cassandra.tcm.transformations.SealPeriod;
import static org.apache.cassandra.distributed.test.log.PlacementSimulator.*;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.*;
import static org.apache.cassandra.distributed.test.log.PlacementSimulator.SimulatedPlacements;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.Node;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.NtsReplicationFactor;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.ReplicationFactor;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.SimpleReplicationFactor;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.nodeFactory;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.nodeFactoryHumanReadable;
public class MetadataChangeSimulationTest extends CMSTestBase
{
@ -87,9 +92,7 @@ public class MetadataChangeSimulationTest extends CMSTestBase
for (int concurrency : new int[]{ 1, 3, 5 })
{
for (int rf : new int[]{ 2, 3, 5 })
{
simulate(50, new NtsReplicationFactor(3, rf), concurrency);
}
simulate(50, 0, new NtsReplicationFactor(3, rf), concurrency);
}
}
@ -99,9 +102,7 @@ public class MetadataChangeSimulationTest extends CMSTestBase
for (int concurrency : new int[]{ 1, 3, 5 })
{
for (int rf : new int[]{ 2, 3, 5 })
{
simulate(50, new SimpleReplicationFactor(rf), concurrency);
}
simulate(50, 0, new SimpleReplicationFactor(rf), concurrency);
}
}
@ -181,11 +182,15 @@ public class MetadataChangeSimulationTest extends CMSTestBase
@Test
public void testLeaveReal() throws Throwable
{
testLeaveReal(new NtsReplicationFactor(1, 3), 1);
testLeaveReal(new NtsReplicationFactor(1, 3), 5);
testLeaveReal(new NtsReplicationFactor(3, 3), 1);
testLeaveReal(new NtsReplicationFactor(3, 3), 5);
for (int i = 1; i <= 12; i++)
{
testLeaveReal(new SimpleReplicationFactor(3), i);
testLeaveReal(new NtsReplicationFactor(1, 3), i);
testLeaveReal(new NtsReplicationFactor(1, 3), i);
testLeaveReal(new NtsReplicationFactor(3, 3), i);
testLeaveReal(new NtsReplicationFactor(3, 3), i);
testLeaveReal(new NtsReplicationFactor(3, 3), i);
}
}
public void testLeaveReal(ReplicationFactor rf, int decomNodeId) throws Throwable
@ -204,6 +209,7 @@ public class MetadataChangeSimulationTest extends CMSTestBase
state = SimulatedOperation.joinWithoutBootstrap(registration.l, sut, registration.r);
}
validatePlacements(sut, state);
state = SimulatedOperation.leave(sut, state, decomNode);
while (!state.inFlightOperations.isEmpty())
@ -214,11 +220,80 @@ public class MetadataChangeSimulationTest extends CMSTestBase
}
}
@Test
public void wraparoundStressTest() throws Throwable
{
wraparoundStressTest(new SimpleReplicationFactor(3));
}
public void wraparoundStressTest(ReplicationFactor rf) throws Throwable
{
try (CMSTestBase.CMSSut sut = new CMSTestBase.CMSSut(AtomicLongBackedProcessor::new, false, rf))
{
ModelState state = ModelState.empty(nodeFactoryHumanReadable(), 10, 1);
Random rng = new Random(1l);
for (int i = 0; i < 20; i++)
{
long token = rng.nextLong();
ModelChecker.Pair<ModelState, Node> registration = registerNewNodeWithToken(state, sut, token, 1, 1);
state = SimulatedOperation.joinWithoutBootstrap(registration.l, sut, registration.r);
}
validatePlacements(sut, state);
ModelChecker.Pair<ModelState, Node> res = registerNewNodeWithToken(state, sut, Long.MIN_VALUE, 1, 1);
Node minTokenNode = res.r;
state = res.l;
boolean isMinJoined = state.currentNodes.stream().anyMatch(n -> n.token() == Long.MIN_VALUE);
for (int i = 0; i < 100; i++)
{
boolean joiningMin = !isMinJoined;
if (joiningMin)
state = SimulatedOperation.join(sut, state, minTokenNode);
else
state = SimulatedOperation.leave(sut, state, minTokenNode);
// Join one more node
if (rng.nextBoolean())
{
res = registerNewNodeWithToken(state, sut, rng.nextLong(), 1, 1);
Node newNode = res.r;
state = res.l;
state = SimulatedOperation.join(sut, state, newNode);
while (!state.inFlightOperations.isEmpty())
{
state = state.inFlightOperations.get(state.inFlightOperations.size() - 1).advance(state);
validatePlacements(sut, state);
}
}
else
{
Node leavingNode = null;
while (leavingNode == null)
{
Node toLeave = state.currentNodes.get(rng.nextInt(state.currentNodes.size() - 1));
if (toLeave.token() != minTokenNode.token())
leavingNode = toLeave;
}
state = SimulatedOperation.leave(sut, state, leavingNode);
while (!state.inFlightOperations.isEmpty())
{
state = state.inFlightOperations.get(state.inFlightOperations.size() - 1).advance(state);
validatePlacements(sut, state);
}
}
isMinJoined = joiningMin;
}
}
}
@Test
public void testJoinReal() throws Throwable
{
testJoinReal(new NtsReplicationFactor(3, 3), 1);
testJoinReal(new NtsReplicationFactor(3, 3), 5);
testJoinReal(new SimpleReplicationFactor(3), 10);
testJoinReal(new NtsReplicationFactor(3, 3), 10);
}
public void testJoinReal(ReplicationFactor rf, int joinNodeId) throws Throwable
@ -250,7 +325,41 @@ public class MetadataChangeSimulationTest extends CMSTestBase
}
}
public void simulate(int toBootstrap, ReplicationFactor rf, int concurrency) throws Throwable
@Test
public void testReplaceReal() throws Throwable
{
testReplaceReal(new SimpleReplicationFactor(3), 10);
testReplaceReal(new NtsReplicationFactor(3, 3), 10);
}
public void testReplaceReal(ReplicationFactor rf, int replacementId) throws Throwable
{
try (CMSTestBase.CMSSut sut = new CMSTestBase.CMSSut(AtomicLongBackedProcessor::new, false, rf))
{
ModelState state = ModelState.empty(nodeFactoryHumanReadable(), 10, 1);
Node toReplace = null;
for (int i = 1; i <= 12; i++)
{
int dc = (i % rf.dcs()) + 1;
ModelChecker.Pair<ModelState, Node> registration = registerNewNode(state, sut, dc, 1);
state = SimulatedOperation.joinWithoutBootstrap(registration.l, sut, registration.r);
if (replacementId == i)
toReplace = registration.r;
}
ModelChecker.Pair<ModelState, Node> replacement = registerNewNode(state, sut, toReplace.tokenIdx(), toReplace.dcIdx(), toReplace.rackIdx());;
state = SimulatedOperation.replace(sut, replacement.l, toReplace, replacement.r);
while (!state.inFlightOperations.isEmpty())
{
state = state.inFlightOperations.get(0).advance(state);
validatePlacements(sut, state);
}
}
}
public void simulate(int toBootstrap, int minSteps, ReplicationFactor rf, int concurrency) throws Throwable
{
System.out.printf("RUNNING SIMULATION. TO BOOTSTRAP: %s, RF: %s, CONCURRENCY: %s%n",
toBootstrap, rf, concurrency);
@ -371,7 +480,7 @@ public class MetadataChangeSimulationTest extends CMSTestBase
return false;
})
.run();
.run(minSteps, Integer.MAX_VALUE);
}
@Test
@ -391,7 +500,6 @@ public class MetadataChangeSimulationTest extends CMSTestBase
public void simulateBounces(ReplicationFactor rf, CMSPlacementStrategy CMSConfigurationStrategy, Random random) throws Throwable
{
try(CMSSut sut = new CMSSut(AtomicLongBackedProcessor::new, false, rf))
{
ModelState state = ModelState.empty(nodeFactory(), 300, 1);
@ -416,7 +524,6 @@ public class MetadataChangeSimulationTest extends CMSTestBase
{
Set<NodeId> bouncing = new HashSet<>();
Set<NodeId> replicasFromBouncedReplicaSets = new HashSet<>();
int j = 0;
outer:
for (VersionedEndpoints.ForRange placements : sut.service.metadata().placements.get(rf.asKeyspaceParams().replication).writes.replicaGroups().values())
{
@ -436,7 +543,6 @@ public class MetadataChangeSimulationTest extends CMSTestBase
bouncing.add(toBounce);
replicasFromBouncedReplicaSets.addAll(replicas);
}
j++;
}
int majority = newCms.size() / 2 + 1;
@ -499,6 +605,14 @@ public class MetadataChangeSimulationTest extends CMSTestBase
return pair(newState, node);
}
private ModelChecker.Pair<ModelState, Node> registerNewNodeWithToken(ModelState state, CMSSut sut, long token, int dcIdx, int rackIdx)
{
ModelState newState = state.transformer().incrementUniqueNodes().transform();
Node node = state.nodeFactory.make(newState.uniqueNodes, dcIdx, rackIdx).overrideToken(token);
sut.service.commit(new Register(new NodeAddresses(node.addr()), new Location(node.dc(), node.rack()), NodeVersion.CURRENT));
return pair(newState, node);
}
private Node getRemovalCandidate(ModelState state, ModelChecker.EntropySource entropySource)
{
return getCandidate(state, entropySource);

6
test/distributed/org/apache/cassandra/distributed/test/log/ModelChecker.java
View File

@ -39,10 +39,10 @@ public class ModelChecker<STATE, SUT>
public void run() throws Throwable
{
run(Integer.MAX_VALUE);
run(0, Integer.MAX_VALUE);
}
public void run(int maxSteps) throws Throwable
public void run(int minSteps, int maxSteps) throws Throwable
{
assert exitCondition != null : "Exit condition is not specified";
assert init != null : "Initial condition is not specified";
@ -54,7 +54,7 @@ public class ModelChecker<STATE, SUT>
for (int i = 0; i < maxSteps; i++)
{
if (exitCondition.test(state.get()))
if (i > minSteps && exitCondition.test(state.get()))
return;
// TODO: add randomisation / probability for triggering a specific step

38
test/distributed/org/apache/cassandra/distributed/test/log/PlacementSimulator.java
View File

@ -43,6 +43,7 @@ import org.junit.Assert;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.Node;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.Range;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.ReplicationFactor;
import static org.apache.cassandra.harry.sut.TokenPlacementModel.toRanges;
/**
* A small class that helps to avoid doing mental arithmetics on ranges.
@ -515,8 +516,8 @@ public class PlacementSimulator
afterLeaveNodes.remove(toRemove);
// calculate placements based on existing ranges but final set of nodes - this is end state
Map<Range, List<Node>> end = baseState.rf.replicate(toRanges(baseState.nodes), afterLeaveNodes).placementsForRange;
// maximal state is union of start & end
// maximal state is union of start & end
Map<Range, Diff<Node>> allWriteCommands = diff(start, end);
Map<Range, Diff<Node>> step1WriteCommands = map(allWriteCommands, Diff::onlyAdditions);
Map<Range, Diff<Node>> step3WriteCommands = map(allWriteCommands, Diff::onlyRemovals);
@ -829,6 +830,11 @@ public class PlacementSimulator
public static NavigableMap<Range, List<Node>> mergeReplicated(Map<Range, List<Node>> orig, long removingToken)
{
if (removingToken == Long.MIN_VALUE)
{
Assert.assertEquals(Long.MIN_VALUE, orig.entrySet().iterator().next().getKey().start);
return new TreeMap<>(orig);
}
NavigableMap<Range, List<Node>> newState = new TreeMap<>();
Iterator<Map.Entry<Range, List<Node>>> iter = orig.entrySet().iterator();
while (iter.hasNext())
@ -854,6 +860,11 @@ public class PlacementSimulator
public static NavigableMap<Range, List<Node>> splitReplicated(Map<Range, List<Node>> orig, long splitAt)
{
if (splitAt == Long.MIN_VALUE)
{
Assert.assertEquals(Long.MIN_VALUE, orig.entrySet().iterator().next().getKey().start);
return new TreeMap<>(orig);
}
NavigableMap<Range, List<Node>> newState = new TreeMap<>();
for (Map.Entry<Range, List<Node>> entry : orig.entrySet())
{
@ -937,31 +948,6 @@ public class PlacementSimulator
return Collections.unmodifiableList(newNodes);
}
/**
* Generates token ranges from the list of nodes
*/
public static Range[] toRanges(List<Node> nodes)
{
List<Long> tokens = new ArrayList<>();
for (Node node : nodes)
tokens.add(node.token());
tokens.add(Long.MIN_VALUE);
tokens.sort(Long::compareTo);
Range[] ranges = new Range[nodes.size() + 1];
long prev = tokens.get(0);
int cnt = 0;
for (int i = 1; i < tokens.size(); i++)
{
long current = tokens.get(i);
ranges[cnt++] = new Range(prev, current);
prev = current;
}
ranges[ranges.length - 1] = new Range(prev, Long.MIN_VALUE);
return ranges;
}
public static class Diff<T> {
public final List<T> additions;
public final List<T> removals;

29
test/distributed/org/apache/cassandra/distributed/test/log/PlacementSimulatorTest.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.distributed.test.log;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.PrimitiveIterator;
@ -55,7 +56,13 @@ public class PlacementSimulatorTest
@Test
public void testMove()
{
testMove(100, 200, 300, 400, 350, new SimpleReplicationFactor(3));
testMove(100);
testMove(Long.MIN_VALUE);
}
public void testMove(long minToken)
{
testMove(minToken, 200, 300, 400, 350, new SimpleReplicationFactor(3));
Random rng = new Random();
for (int i = 0; i < 1000; i++)
@ -108,7 +115,13 @@ public class PlacementSimulatorTest
@Test
public void testBootstrap()
{
testBootstrap(100, 200, 300, 400, 350, new SimpleReplicationFactor(3));
testBootstrap(350);
testBootstrap(Long.MIN_VALUE);
}
public void testBootstrap(long newToken)
{
testBootstrap(100, 200, 300, 400, newToken, new SimpleReplicationFactor(3));
Random rng = new Random();
for (int i = 0; i < 1000; i++)
@ -162,7 +175,13 @@ public class PlacementSimulatorTest
@Test
public void testDecommission()
{
testDecommission(100, 200, 300, 400, 350, new SimpleReplicationFactor(3));
testDecommission(100);
testDecommission(Long.MIN_VALUE);
}
public void testDecommission(long minToken)
{
testDecommission(minToken, 200, 300, 400, 350, new SimpleReplicationFactor(3));
Random rng = new Random();
for (int i = 0; i < 1000; i++)
@ -308,6 +327,7 @@ public class PlacementSimulatorTest
List<Node> nodes = new ArrayList<>(10);
for (int i = 1; i <= 10; i++)
nodes.add(factory.make(i, 1, 1));
nodes.sort(Comparator.comparing(Node::token));
SimulatedPlacements sim = new SimulatedPlacements(rf, nodes, rf.replicate(nodes).asMap(), rf.replicate(nodes).asMap(), Collections.emptyList());
Node newNode = factory.make(11, 1, 1);
@ -325,7 +345,7 @@ public class PlacementSimulatorTest
List<Node> nodes = new ArrayList<>(10);
for (int i = 1; i <= 10; i++)
nodes.add(factory.make(i, 1, 1));
nodes.sort(Comparator.comparing(Node::token));
Node toRemove = nodes.get(5);
SimulatedPlacements sim = new SimulatedPlacements(rf, nodes, rf.replicate(nodes).asMap(), rf.replicate(nodes).asMap(), Collections.emptyList());
revertPartiallyCompleteOp(sim, () -> leave(sim, toRemove), 2);
@ -342,6 +362,7 @@ public class PlacementSimulatorTest
List<Node> nodes = new ArrayList<>(10);
for (int i = 1; i <= 10; i++)
nodes.add(factory.make(i, 1, 1));
nodes.sort(Comparator.comparing(Node::token));
Node toReplace = nodes.get(5);
SimulatedPlacements sim = new SimulatedPlacements(rf,

1
test/distributed/org/apache/cassandra/distributed/test/log/SimulatedOperation.java
View File

@ -110,6 +110,7 @@ public abstract class SimulatedOperation
public static ModelState replace(CMSSut sut, ModelState state, Node toReplace, Node replacement)
{
assert toReplace.tokenIdx() == replacement.tokenIdx();
ModelState.Transformer transformer = state.transformer();
new Replace(toReplace, replacement).create(sut, state.simulatedPlacements, transformer);
return transformer.transform();

124
test/harry/main/org/apache/cassandra/harry/sut/TokenPlacementModel.java
View File

@ -58,6 +58,7 @@ public class TokenPlacementModel
{
return replicate(toRanges(nodes), nodes);
}
public abstract ReplicatedRanges replicate(Range[] ranges, List<Node> nodes);
}
@ -79,7 +80,7 @@ public class TokenPlacementModel
if (token <= range.start)
return 1;
// ie token > start && token <= end
if (token <= range.end ||range.end == Long.MIN_VALUE)
if (token <= range.end || range.end == Long.MIN_VALUE)
return 0;
return -1;
@ -135,7 +136,13 @@ public class TokenPlacementModel
{
for (int i = 0; i < nodes.size(); i++)
{
if (range.end != Long.MIN_VALUE && nodes.get(i).token() >= range.end)
long token = nodes.get(i).token();
if (token == Long.MIN_VALUE)
{
if (range.end == token)
return i;
}
else if (range.end != Long.MIN_VALUE && token >= range.end)
return i;
}
return -1;
@ -147,13 +154,15 @@ public class TokenPlacementModel
*/
public static Range[] toRanges(List<Node> nodes)
{
boolean hasMinToken = nodes.get(0).token() == Long.MIN_VALUE;
List<Long> tokens = new ArrayList<>();
for (Node node : nodes)
tokens.add(node.token());
tokens.add(Long.MIN_VALUE);
if (!hasMinToken)
tokens.add(Long.MIN_VALUE);
tokens.sort(Long::compareTo);
Range[] ranges = new Range[nodes.size() + 1];
Range[] ranges = new Range[nodes.size() + (hasMinToken ? 0 : 1)];
long prev = tokens.get(0);
int cnt = 0;
for (int i = 1; i < tokens.size(); i++)
@ -244,23 +253,10 @@ public class TokenPlacementModel
return replicate(ranges, nodes, asMap());
}
private static <T extends Comparable<T>> void assertStrictlySorted(Collection<T> coll)
{
if (coll.size() <= 1) return;
Iterator<T> iter = coll.iterator();
T prev = iter.next();
while (iter.hasNext())
{
T next = iter.next();
assert next.compareTo(prev) > 0 : String.format("Collection does not seem to be sorted. %s and %s are in wrong order", prev, next);
prev = next;
}
}
public static ReplicatedRanges replicate(Range[] ranges, List<Node> nodes, Map<String, Integer> rfs)
{
assertStrictlySorted(nodes);
boolean minTokenOwned = nodes.stream().anyMatch(n -> n.token() == Long.MIN_VALUE);
Map<String, DatacenterNodes> template = new HashMap<>();
Map<String, List<Node>> nodesByDC = nodesByDC(nodes);
@ -281,7 +277,7 @@ public class TokenPlacementModel
}
NavigableMap<Range, Map<String, List<Node>>> replication = new TreeMap<>();
Range skipped = null;
for (Range range : ranges)
{
final int idx = primaryReplica(nodes, range);
@ -308,14 +304,21 @@ public class TokenPlacementModel
}
else
{
// if the range end is larger than the highest assigned token, then treat it
// as part of the wraparound and replicate it to the same nodes as the first
// range. This is most likely caused by a decommission removing the node with
// the largest token.
replication.put(range, replication.get(ranges[0]));
if (minTokenOwned)
skipped = range;
else
// if the range end is larger than the highest assigned token, then treat it
// as part of the wraparound and replicate it to the same nodes as the first
// range. This is most likely caused by a decommission removing the node with
// the largest token.
replication.put(range, replication.get(ranges[0]));
}
}
// Since we allow owning MIN_TOKEN, when it is owned, we have to replicate the range explicitly.
if (skipped != null)
replication.put(skipped, replication.get(ranges[ranges.length - 1]));
return combine(replication);
}
@ -446,6 +449,20 @@ public class TokenPlacementModel
}
}
private static <T extends Comparable<T>> void assertStrictlySorted(Collection<T> coll)
{
if (coll.size() <= 1) return;
Iterator<T> iter = coll.iterator();
T prev = iter.next();
while (iter.hasNext())
{
T next = iter.next();
assert next.compareTo(prev) > 0 : String.format("Collection does not seem to be sorted. %s and %s are in wrong order", prev, next);
prev = next;
}
}
private static <K extends Comparable<K>, T1, T2> Map<K, T2> mapValues(Map<K, T1> allDCs, Function<T1, T2> map)
{
NavigableMap<K, T2> res = new TreeMap<>();
@ -515,7 +532,10 @@ public class TokenPlacementModel
public static ReplicatedRanges replicate(Range[] ranges, List<Node> nodes, int rf)
{
assertStrictlySorted(nodes);
NavigableMap<Range, List<Node>> replication = new TreeMap<>();
boolean minTokenOwned = nodes.stream().anyMatch(n -> n.token() == Long.MIN_VALUE);
Range skipped = null;
for (Range range : ranges)
{
Set<Integer> names = new HashSet<>();
@ -523,25 +543,29 @@ public class TokenPlacementModel
int idx = primaryReplica(nodes, range);
if (idx >= 0)
{
for (int i = idx; i < nodes.size() && replicas.size() < rf; i++)
addIfUnique(replicas, names, nodes.get(i));
for (int i = 0; replicas.size() < rf && i < idx; i++)
addIfUnique(replicas, names, nodes.get(i));
if (range.start == Long.MIN_VALUE)
for (int i = 0; i < nodes.size() && replicas.size() < rf; i++)
addIfUnique(replicas, names, nodes.get((idx + i) % nodes.size()));
if (!minTokenOwned && range.start == Long.MIN_VALUE)
replication.put(ranges[ranges.length - 1], replicas);
replication.put(range, replicas);
}
else
{
// if the range end is larger than the highest assigned token, then treat it
// as part of the wraparound and replicate it to the same nodes as the first
// range. This is most likely caused by a decommission removing the node with
// the largest token.
replication.put(range, replication.get(ranges[0]));
if (minTokenOwned)
skipped = range;
else
// if the range end is larger than the highest assigned token, then treat it
// as part of the wraparound and replicate it to the same nodes as the first
// range. This is most likely caused by a decommission removing the node with
// the largest token.
replication.put(range, replication.get(ranges[0]));
}
}
// Since we allow owning MIN_TOKEN, when it is owned, we have to replicate the range explicitly.
if (skipped != null)
replication.put(skipped, replication.get(ranges[ranges.length - 1]));
return new ReplicatedRanges(ranges, Collections.unmodifiableNavigableMap(replication));
}
@ -686,7 +710,13 @@ public class TokenPlacementModel
public static class DefaultLookup implements Lookup
{
protected final Map<Integer, Long> overrides = new HashMap<>(2);
protected final Map<Integer, Long> tokenOverrides = new HashMap<>(2);
public DefaultLookup()
{
// A crafty way to introduce a MIN token
tokenOverrides.put(10, Long.MIN_VALUE);
}
public String id(int nodeIdx)
{
@ -695,23 +725,29 @@ public class TokenPlacementModel
public long token(int tokenIdx)
{
Long override = overrides.get(tokenIdx);
Long override = tokenOverrides.get(tokenIdx);
if (override != null)
return override;
return PCGFastPure.next(tokenIdx, 1L);
long token = PCGFastPure.next(tokenIdx, 1L);
for (Long value : tokenOverrides.values())
{
if (token == value)
throw new IllegalStateException(String.format("Generated token %d is already used in an override", token));
}
return token;
}
public Lookup forceToken(int tokenIdx, long token)
{
DefaultLookup newLookup = new DefaultLookup();
newLookup.overrides.putAll(overrides);
newLookup.overrides.put(tokenIdx, token);
newLookup.tokenOverrides.putAll(tokenOverrides);
newLookup.tokenOverrides.put(tokenIdx, token);
return newLookup;
}
public void reset()
{
overrides.clear();
tokenOverrides.clear();
}
public String dc(int dcIdx)
@ -729,7 +765,7 @@ public class TokenPlacementModel
@Override
public long token(int tokenIdx)
{
Long override = overrides.get(tokenIdx);
Long override = tokenOverrides.get(tokenIdx);
if (override != null)
return override;
return tokenIdx * 100L;
@ -738,8 +774,8 @@ public class TokenPlacementModel
public Lookup forceToken(int tokenIdx, long token)
{
DefaultLookup lookup = new HumanReadableTokensLookup();
lookup.overrides.putAll(overrides);
lookup.overrides.put(tokenIdx, token);
lookup.tokenOverrides.putAll(tokenOverrides);
lookup.tokenOverrides.put(tokenIdx, token);
return lookup;
}
}

4
test/simulator/main/org/apache/cassandra/simulator/cluster/KeyspaceActions.java
View File

@ -428,8 +428,8 @@ public class KeyspaceActions extends ClusterActions
public TokenPlacementModel.Lookup forceToken(int tokenIdx, long token)
{
SimulationLookup newLookup = new SimulationLookup();
newLookup.overrides.putAll(overrides);
newLookup.overrides.put(tokenIdx, token);
newLookup.tokenOverrides.putAll(tokenOverrides);
newLookup.tokenOverrides.put(tokenIdx, token);
return newLookup;
}
}