mirror of https://github.com/apache/cassandra
Remove obsolete OldNetworkTopologyStrategy
Removed the strategy from cqlsh autocomplete, including an array for replication_factor autocomplete that was only used for SimpleStrategy and OldNetworkTopologyStrategy. patch by Pedro Gordo; reviewed by Anthony Grasso, Mick Semb Wever for CASSANDRA-13990
This commit is contained in:
Pedro Gordo
Mick Semb Wever
parent
5eea8bf7b5
commit
7c5904753f
|
|
@ -77,3 +77,5 @@ lib/jsr223/scala/*.jar
|
||||||
doc/source/configuration/cassandra_config_file.rst
|
doc/source/configuration/cassandra_config_file.rst
|
||||||
doc/source/tools/nodetool
|
doc/source/tools/nodetool
|
||||||
|
|
||||||
|
# Python virtual environment
|
||||||
|
venv/
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,5 @@
|
||||||
4.0-alpha3
|
4.0-alpha3
|
||||||
|
* Removed obsolete OldNetworkTopologyStrategy (CASSANDRA-13990)
|
||||||
* Align record header of FQL and audit binary log (CASSANDRA-15076)
|
* Align record header of FQL and audit binary log (CASSANDRA-15076)
|
||||||
* Shuffle forwarding replica for messages to non-local DC (CASSANDRA-15318)
|
* Shuffle forwarding replica for messages to non-local DC (CASSANDRA-15318)
|
||||||
* Optimise native protocol ASCII string encoding (CASSANDRA-15410)
|
* Optimise native protocol ASCII string encoding (CASSANDRA-15410)
|
||||||
|
|
|
||||||
|
|
@ -175,7 +175,6 @@ The supported @<properties>@ for @CREATE KEYSPACE@ are:
|
||||||
The @replication@ @<property>@ is mandatory. It must at least contains the @'class'@ sub-option which defines the replication strategy class to use. The rest of the sub-options depends on that replication strategy class. By default, Cassandra support the following @'class'@:
|
The @replication@ @<property>@ is mandatory. It must at least contains the @'class'@ sub-option which defines the replication strategy class to use. The rest of the sub-options depends on that replication strategy class. By default, Cassandra support the following @'class'@:
|
||||||
* @'SimpleStrategy'@: A simple strategy that defines a simple replication factor for the whole cluster. The only sub-options supported is @'replication_factor'@ to define that replication factor and is mandatory.
|
* @'SimpleStrategy'@: A simple strategy that defines a simple replication factor for the whole cluster. The only sub-options supported is @'replication_factor'@ to define that replication factor and is mandatory.
|
||||||
* @'NetworkTopologyStrategy'@: A replication strategy that allows to set the replication factor independently for each data-center. The rest of the sub-options are key-value pairs where each time the key is the name of a datacenter and the value the replication factor for that data-center.
|
* @'NetworkTopologyStrategy'@: A replication strategy that allows to set the replication factor independently for each data-center. The rest of the sub-options are key-value pairs where each time the key is the name of a datacenter and the value the replication factor for that data-center.
|
||||||
* @'OldNetworkTopologyStrategy'@: A legacy replication strategy. You should avoid this strategy for new keyspaces and prefer @'NetworkTopologyStrategy'@.
|
|
||||||
|
|
||||||
Attempting to create an already existing keyspace will return an error unless the @IF NOT EXISTS@ option is used. If it is used, the statement will be a no-op if the keyspace already exists.
|
Attempting to create an already existing keyspace will return an error unless the @IF NOT EXISTS@ option is used. If it is used, the statement will be a no-op if the keyspace already exists.
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -450,7 +450,7 @@ def ks_prop_val_mapkey_completer(ctxt, cass):
|
||||||
break
|
break
|
||||||
else:
|
else:
|
||||||
return ["'class'"]
|
return ["'class'"]
|
||||||
if repclass in CqlRuleSet.replication_factor_strategies:
|
if repclass == 'SimpleStrategy':
|
||||||
opts = set(('replication_factor',))
|
opts = set(('replication_factor',))
|
||||||
elif repclass == 'NetworkTopologyStrategy':
|
elif repclass == 'NetworkTopologyStrategy':
|
||||||
return [Hint('<dc_name>')]
|
return [Hint('<dc_name>')]
|
||||||
|
|
@ -479,7 +479,7 @@ def ks_prop_val_mapender_completer(ctxt, cass):
|
||||||
break
|
break
|
||||||
else:
|
else:
|
||||||
return [',']
|
return [',']
|
||||||
if repclass in CqlRuleSet.replication_factor_strategies:
|
if repclass == 'SimpleStrategy':
|
||||||
if 'replication_factor' not in keysseen:
|
if 'replication_factor' not in keysseen:
|
||||||
return [',']
|
return [',']
|
||||||
if repclass == 'NetworkTopologyStrategy' and len(keysseen) == 1:
|
if repclass == 'NetworkTopologyStrategy' and len(keysseen) == 1:
|
||||||
|
|
|
||||||
|
|
@ -41,17 +41,9 @@ class CqlParsingRuleSet(pylexotron.ParsingRuleSet):
|
||||||
|
|
||||||
replication_strategies = (
|
replication_strategies = (
|
||||||
'SimpleStrategy',
|
'SimpleStrategy',
|
||||||
'OldNetworkTopologyStrategy',
|
|
||||||
'NetworkTopologyStrategy'
|
'NetworkTopologyStrategy'
|
||||||
)
|
)
|
||||||
|
|
||||||
replication_factor_strategies = (
|
|
||||||
'SimpleStrategy',
|
|
||||||
'org.apache.cassandra.locator.SimpleStrategy',
|
|
||||||
'OldNetworkTopologyStrategy',
|
|
||||||
'org.apache.cassandra.locator.OldNetworkTopologyStrategy'
|
|
||||||
)
|
|
||||||
|
|
||||||
def __init__(self, *args, **kwargs):
|
def __init__(self, *args, **kwargs):
|
||||||
pylexotron.ParsingRuleSet.__init__(self, *args, **kwargs)
|
pylexotron.ParsingRuleSet.__init__(self, *args, **kwargs)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,127 +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.locator;
|
|
||||||
|
|
||||||
import java.util.ArrayList;
|
|
||||||
import java.util.Collections;
|
|
||||||
import java.util.Collection;
|
|
||||||
import java.util.Iterator;
|
|
||||||
import java.util.Map;
|
|
||||||
|
|
||||||
import org.apache.cassandra.dht.Range;
|
|
||||||
import org.apache.cassandra.exceptions.ConfigurationException;
|
|
||||||
import org.apache.cassandra.dht.Token;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* This Replication Strategy returns the nodes responsible for a given
|
|
||||||
* key but respects rack awareness. It places one replica in a
|
|
||||||
* different data center from the first (if there is any such data center),
|
|
||||||
* the third replica in a different rack in the first datacenter, and
|
|
||||||
* any remaining replicas on the first unused nodes on the ring.
|
|
||||||
*/
|
|
||||||
public class OldNetworkTopologyStrategy extends AbstractReplicationStrategy
|
|
||||||
{
|
|
||||||
private final ReplicationFactor rf;
|
|
||||||
public OldNetworkTopologyStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
|
|
||||||
{
|
|
||||||
super(keyspaceName, tokenMetadata, snitch, configOptions);
|
|
||||||
this.rf = ReplicationFactor.fromString(this.configOptions.get("replication_factor"));
|
|
||||||
}
|
|
||||||
|
|
||||||
public EndpointsForRange calculateNaturalReplicas(Token token, TokenMetadata metadata)
|
|
||||||
{
|
|
||||||
ArrayList<Token> tokens = metadata.sortedTokens();
|
|
||||||
if (tokens.isEmpty())
|
|
||||||
return EndpointsForRange.empty(new Range<>(metadata.partitioner.getMinimumToken(), metadata.partitioner.getMinimumToken()));
|
|
||||||
|
|
||||||
Iterator<Token> iter = TokenMetadata.ringIterator(tokens, token, false);
|
|
||||||
Token primaryToken = iter.next();
|
|
||||||
Token previousToken = metadata.getPredecessor(primaryToken);
|
|
||||||
Range<Token> tokenRange = new Range<>(previousToken, primaryToken);
|
|
||||||
|
|
||||||
EndpointsForRange.Builder replicas = new EndpointsForRange.Builder(tokenRange, rf.allReplicas);
|
|
||||||
|
|
||||||
assert !rf.hasTransientReplicas() : "support transient replicas";
|
|
||||||
replicas.add(new Replica(metadata.getEndpoint(primaryToken), previousToken, primaryToken, true));
|
|
||||||
|
|
||||||
boolean bDataCenter = false;
|
|
||||||
boolean bOtherRack = false;
|
|
||||||
while (replicas.size() < rf.allReplicas && iter.hasNext())
|
|
||||||
{
|
|
||||||
// First try to find one in a different data center
|
|
||||||
Token t = iter.next();
|
|
||||||
if (!snitch.getDatacenter(metadata.getEndpoint(primaryToken)).equals(snitch.getDatacenter(metadata.getEndpoint(t))))
|
|
||||||
{
|
|
||||||
// If we have already found something in a diff datacenter no need to find another
|
|
||||||
if (!bDataCenter)
|
|
||||||
{
|
|
||||||
replicas.add(new Replica(metadata.getEndpoint(t), previousToken, primaryToken, true));
|
|
||||||
bDataCenter = true;
|
|
||||||
}
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
// Now try to find one on a different rack
|
|
||||||
if (!snitch.getRack(metadata.getEndpoint(primaryToken)).equals(snitch.getRack(metadata.getEndpoint(t))) &&
|
|
||||||
snitch.getDatacenter(metadata.getEndpoint(primaryToken)).equals(snitch.getDatacenter(metadata.getEndpoint(t))))
|
|
||||||
{
|
|
||||||
// If we have already found something in a diff rack no need to find another
|
|
||||||
if (!bOtherRack)
|
|
||||||
{
|
|
||||||
replicas.add(new Replica(metadata.getEndpoint(t), previousToken, primaryToken, true));
|
|
||||||
bOtherRack = true;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
// If we found N number of nodes we are good. This loop wil just exit. Otherwise just
|
|
||||||
// loop through the list and add until we have N nodes.
|
|
||||||
if (replicas.size() < rf.allReplicas)
|
|
||||||
{
|
|
||||||
iter = TokenMetadata.ringIterator(tokens, token, false);
|
|
||||||
while (replicas.size() < rf.allReplicas && iter.hasNext())
|
|
||||||
{
|
|
||||||
Token t = iter.next();
|
|
||||||
Replica replica = new Replica(metadata.getEndpoint(t), previousToken, primaryToken, true);
|
|
||||||
if (!replicas.endpoints().contains(replica.endpoint()))
|
|
||||||
replicas.add(replica);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return replicas.build();
|
|
||||||
}
|
|
||||||
|
|
||||||
public ReplicationFactor getReplicationFactor()
|
|
||||||
{
|
|
||||||
return rf;
|
|
||||||
}
|
|
||||||
|
|
||||||
public void validateOptions() throws ConfigurationException
|
|
||||||
{
|
|
||||||
if (configOptions == null || configOptions.get("replication_factor") == null)
|
|
||||||
{
|
|
||||||
throw new ConfigurationException("SimpleStrategy requires a replication_factor strategy option.");
|
|
||||||
}
|
|
||||||
validateReplicationFactor(configOptions.get("replication_factor"));
|
|
||||||
}
|
|
||||||
|
|
||||||
public Collection<String> recognizedOptions()
|
|
||||||
{
|
|
||||||
return Collections.<String>singleton("replication_factor");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,386 +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.locator;
|
|
||||||
|
|
||||||
import java.net.UnknownHostException;
|
|
||||||
import java.util.ArrayList;
|
|
||||||
import java.util.Arrays;
|
|
||||||
import java.util.Collections;
|
|
||||||
import java.util.HashMap;
|
|
||||||
import java.util.List;
|
|
||||||
import java.util.Map;
|
|
||||||
import java.util.Set;
|
|
||||||
|
|
||||||
import org.junit.Before;
|
|
||||||
import org.junit.BeforeClass;
|
|
||||||
import org.junit.Test;
|
|
||||||
|
|
||||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
|
||||||
import org.apache.cassandra.dht.RandomPartitioner.BigIntegerToken;
|
|
||||||
import org.apache.cassandra.dht.Range;
|
|
||||||
import org.apache.cassandra.dht.Token;
|
|
||||||
import org.apache.cassandra.service.RangeRelocator;
|
|
||||||
import org.apache.cassandra.service.StorageService;
|
|
||||||
import org.apache.cassandra.utils.Pair;
|
|
||||||
|
|
||||||
import static org.junit.Assert.assertEquals;
|
|
||||||
import static org.junit.Assert.assertTrue;
|
|
||||||
|
|
||||||
public class OldNetworkTopologyStrategyTest
|
|
||||||
{
|
|
||||||
|
|
||||||
private List<Token> keyTokens;
|
|
||||||
private TokenMetadata tmd;
|
|
||||||
private Map<String, ArrayList<InetAddressAndPort>> expectedResults;
|
|
||||||
|
|
||||||
@BeforeClass
|
|
||||||
public static void setupDD()
|
|
||||||
{
|
|
||||||
DatabaseDescriptor.daemonInitialization();
|
|
||||||
}
|
|
||||||
|
|
||||||
@Before
|
|
||||||
public void init() throws Exception
|
|
||||||
{
|
|
||||||
keyTokens = new ArrayList<Token>();
|
|
||||||
tmd = new TokenMetadata();
|
|
||||||
expectedResults = new HashMap<String, ArrayList<InetAddressAndPort>>();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 4 same rack endpoints
|
|
||||||
*
|
|
||||||
* @throws java.net.UnknownHostException
|
|
||||||
*/
|
|
||||||
@Test
|
|
||||||
public void testBigIntegerEndpointsA() throws UnknownHostException
|
|
||||||
{
|
|
||||||
RackInferringSnitch endpointSnitch = new RackInferringSnitch();
|
|
||||||
|
|
||||||
AbstractReplicationStrategy strategy = new OldNetworkTopologyStrategy("Keyspace1", tmd, endpointSnitch, optsWithRF(1));
|
|
||||||
addEndpoint("0", "5", "254.0.0.1");
|
|
||||||
addEndpoint("10", "15", "254.0.0.2");
|
|
||||||
addEndpoint("20", "25", "254.0.0.3");
|
|
||||||
addEndpoint("30", "35", "254.0.0.4");
|
|
||||||
|
|
||||||
expectedResults.put("5", buildResult("254.0.0.2", "254.0.0.3", "254.0.0.4"));
|
|
||||||
expectedResults.put("15", buildResult("254.0.0.3", "254.0.0.4", "254.0.0.1"));
|
|
||||||
expectedResults.put("25", buildResult("254.0.0.4", "254.0.0.1", "254.0.0.2"));
|
|
||||||
expectedResults.put("35", buildResult("254.0.0.1", "254.0.0.2", "254.0.0.3"));
|
|
||||||
|
|
||||||
testGetEndpoints(strategy, keyTokens.toArray(new Token[0]));
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 3 same rack endpoints
|
|
||||||
* 1 external datacenter
|
|
||||||
*
|
|
||||||
* @throws java.net.UnknownHostException
|
|
||||||
*/
|
|
||||||
@Test
|
|
||||||
public void testBigIntegerEndpointsB() throws UnknownHostException
|
|
||||||
{
|
|
||||||
RackInferringSnitch endpointSnitch = new RackInferringSnitch();
|
|
||||||
|
|
||||||
AbstractReplicationStrategy strategy = new OldNetworkTopologyStrategy("Keyspace1", tmd, endpointSnitch, optsWithRF(1));
|
|
||||||
addEndpoint("0", "5", "254.0.0.1");
|
|
||||||
addEndpoint("10", "15", "254.0.0.2");
|
|
||||||
addEndpoint("20", "25", "254.1.0.3");
|
|
||||||
addEndpoint("30", "35", "254.0.0.4");
|
|
||||||
|
|
||||||
expectedResults.put("5", buildResult("254.0.0.2", "254.1.0.3", "254.0.0.4"));
|
|
||||||
expectedResults.put("15", buildResult("254.1.0.3", "254.0.0.4", "254.0.0.1"));
|
|
||||||
expectedResults.put("25", buildResult("254.0.0.4", "254.1.0.3", "254.0.0.1"));
|
|
||||||
expectedResults.put("35", buildResult("254.0.0.1", "254.1.0.3", "254.0.0.2"));
|
|
||||||
|
|
||||||
testGetEndpoints(strategy, keyTokens.toArray(new Token[0]));
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 2 same rack endpoints
|
|
||||||
* 1 same datacenter, different rack endpoints
|
|
||||||
* 1 external datacenter
|
|
||||||
*
|
|
||||||
* @throws java.net.UnknownHostException
|
|
||||||
*/
|
|
||||||
@Test
|
|
||||||
public void testBigIntegerEndpointsC() throws UnknownHostException
|
|
||||||
{
|
|
||||||
RackInferringSnitch endpointSnitch = new RackInferringSnitch();
|
|
||||||
|
|
||||||
AbstractReplicationStrategy strategy = new OldNetworkTopologyStrategy("Keyspace1", tmd, endpointSnitch, optsWithRF(1));
|
|
||||||
addEndpoint("0", "5", "254.0.0.1");
|
|
||||||
addEndpoint("10", "15", "254.0.0.2");
|
|
||||||
addEndpoint("20", "25", "254.0.1.3");
|
|
||||||
addEndpoint("30", "35", "254.1.0.4");
|
|
||||||
|
|
||||||
expectedResults.put("5", buildResult("254.0.0.2", "254.0.1.3", "254.1.0.4"));
|
|
||||||
expectedResults.put("15", buildResult("254.0.1.3", "254.1.0.4", "254.0.0.1"));
|
|
||||||
expectedResults.put("25", buildResult("254.1.0.4", "254.0.0.1", "254.0.0.2"));
|
|
||||||
expectedResults.put("35", buildResult("254.0.0.1", "254.0.1.3", "254.1.0.4"));
|
|
||||||
|
|
||||||
testGetEndpoints(strategy, keyTokens.toArray(new Token[0]));
|
|
||||||
}
|
|
||||||
|
|
||||||
private ArrayList<InetAddressAndPort> buildResult(String... addresses) throws UnknownHostException
|
|
||||||
{
|
|
||||||
ArrayList<InetAddressAndPort> result = new ArrayList<>();
|
|
||||||
for (String address : addresses)
|
|
||||||
{
|
|
||||||
result.add(InetAddressAndPort.getByName(address));
|
|
||||||
}
|
|
||||||
return result;
|
|
||||||
}
|
|
||||||
|
|
||||||
private void addEndpoint(String endpointTokenID, String keyTokenID, String endpointAddress) throws UnknownHostException
|
|
||||||
{
|
|
||||||
BigIntegerToken endpointToken = new BigIntegerToken(endpointTokenID);
|
|
||||||
|
|
||||||
BigIntegerToken keyToken = new BigIntegerToken(keyTokenID);
|
|
||||||
keyTokens.add(keyToken);
|
|
||||||
|
|
||||||
InetAddressAndPort ep = InetAddressAndPort.getByName(endpointAddress);
|
|
||||||
tmd.updateNormalToken(endpointToken, ep);
|
|
||||||
}
|
|
||||||
|
|
||||||
private void testGetEndpoints(AbstractReplicationStrategy strategy, Token[] keyTokens)
|
|
||||||
{
|
|
||||||
for (Token keyToken : keyTokens)
|
|
||||||
{
|
|
||||||
int j = 0;
|
|
||||||
for (InetAddressAndPort endpoint : strategy.getNaturalReplicasForToken(keyToken).endpoints())
|
|
||||||
{
|
|
||||||
ArrayList<InetAddressAndPort> hostsExpected = expectedResults.get(keyToken.toString());
|
|
||||||
assertEquals(endpoint, hostsExpected.get(j++));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* test basic methods to move a node. For sure, it's not the best place, but it's easy to test
|
|
||||||
*
|
|
||||||
* @throws java.net.UnknownHostException
|
|
||||||
*/
|
|
||||||
@Test
|
|
||||||
public void testMoveLeft() throws UnknownHostException
|
|
||||||
{
|
|
||||||
// Moves to the left : nothing to fetch, last part to stream
|
|
||||||
|
|
||||||
int movingNodeIdx = 1;
|
|
||||||
BigIntegerToken newToken = new BigIntegerToken("21267647932558653966460912964485513216");
|
|
||||||
BigIntegerToken[] tokens = initTokens();
|
|
||||||
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
|
|
||||||
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
|
|
||||||
|
|
||||||
assertEquals(ranges.left.iterator().next().left, tokensAfterMove[movingNodeIdx]);
|
|
||||||
assertEquals(ranges.left.iterator().next().right, tokens[movingNodeIdx]);
|
|
||||||
assertEquals("No data should be fetched", ranges.right.size(), 0);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
public void testMoveRight() throws UnknownHostException
|
|
||||||
{
|
|
||||||
// Moves to the right : last part to fetch, nothing to stream
|
|
||||||
|
|
||||||
int movingNodeIdx = 1;
|
|
||||||
BigIntegerToken newToken = new BigIntegerToken("35267647932558653966460912964485513216");
|
|
||||||
BigIntegerToken[] tokens = initTokens();
|
|
||||||
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
|
|
||||||
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
|
|
||||||
|
|
||||||
assertEquals("No data should be streamed", ranges.left.size(), 0);
|
|
||||||
assertEquals(ranges.right.iterator().next().left, tokens[movingNodeIdx]);
|
|
||||||
assertEquals(ranges.right.iterator().next().right, tokensAfterMove[movingNodeIdx]);
|
|
||||||
}
|
|
||||||
|
|
||||||
@SuppressWarnings("unchecked")
|
|
||||||
@Test
|
|
||||||
public void testMoveMiddleOfRing() throws UnknownHostException
|
|
||||||
{
|
|
||||||
// moves to another position in the middle of the ring : should stream all its data, and fetch all its new data
|
|
||||||
|
|
||||||
int movingNodeIdx = 1;
|
|
||||||
int movingNodeIdxAfterMove = 4;
|
|
||||||
BigIntegerToken newToken = new BigIntegerToken("90070591730234615865843651857942052864");
|
|
||||||
BigIntegerToken[] tokens = initTokens();
|
|
||||||
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
|
|
||||||
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
|
|
||||||
|
|
||||||
// sort the results, so they can be compared
|
|
||||||
Range<Token>[] toStream = ranges.left.toArray(new Range[0]);
|
|
||||||
Range<Token>[] toFetch = ranges.right.toArray(new Range[0]);
|
|
||||||
Arrays.sort(toStream);
|
|
||||||
Arrays.sort(toFetch);
|
|
||||||
|
|
||||||
// build expected ranges
|
|
||||||
Range<Token>[] toStreamExpected = new Range[2];
|
|
||||||
toStreamExpected[0] = new Range<Token>(getToken(movingNodeIdx - 2, tokens), getToken(movingNodeIdx - 1, tokens));
|
|
||||||
toStreamExpected[1] = new Range<Token>(getToken(movingNodeIdx - 1, tokens), getToken(movingNodeIdx, tokens));
|
|
||||||
Arrays.sort(toStreamExpected);
|
|
||||||
Range<Token>[] toFetchExpected = new Range[2];
|
|
||||||
toFetchExpected[0] = new Range<Token>(getToken(movingNodeIdxAfterMove - 1, tokens), getToken(movingNodeIdxAfterMove, tokens));
|
|
||||||
toFetchExpected[1] = new Range<Token>(getToken(movingNodeIdxAfterMove, tokensAfterMove), getToken(movingNodeIdx, tokensAfterMove));
|
|
||||||
Arrays.sort(toFetchExpected);
|
|
||||||
|
|
||||||
assertEquals(Arrays.equals(toStream, toStreamExpected), true);
|
|
||||||
assertEquals(Arrays.equals(toFetch, toFetchExpected), true);
|
|
||||||
}
|
|
||||||
|
|
||||||
@SuppressWarnings("unchecked")
|
|
||||||
@Test
|
|
||||||
public void testMoveAfterNextNeighbors() throws UnknownHostException
|
|
||||||
{
|
|
||||||
// moves after its next neighbor in the ring
|
|
||||||
|
|
||||||
int movingNodeIdx = 1;
|
|
||||||
int movingNodeIdxAfterMove = 2;
|
|
||||||
BigIntegerToken newToken = new BigIntegerToken("52535295865117307932921825928971026432");
|
|
||||||
BigIntegerToken[] tokens = initTokens();
|
|
||||||
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
|
|
||||||
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
|
|
||||||
|
|
||||||
|
|
||||||
// sort the results, so they can be compared
|
|
||||||
Range<Token>[] toStream = ranges.left.toArray(new Range[0]);
|
|
||||||
Range<Token>[] toFetch = ranges.right.toArray(new Range[0]);
|
|
||||||
Arrays.sort(toStream);
|
|
||||||
Arrays.sort(toFetch);
|
|
||||||
|
|
||||||
// build expected ranges
|
|
||||||
Range<Token>[] toStreamExpected = new Range[1];
|
|
||||||
toStreamExpected[0] = new Range<Token>(getToken(movingNodeIdx - 2, tokens), getToken(movingNodeIdx - 1, tokens));
|
|
||||||
Arrays.sort(toStreamExpected);
|
|
||||||
Range<Token>[] toFetchExpected = new Range[2];
|
|
||||||
toFetchExpected[0] = new Range<Token>(getToken(movingNodeIdxAfterMove - 1, tokens), getToken(movingNodeIdxAfterMove, tokens));
|
|
||||||
toFetchExpected[1] = new Range<Token>(getToken(movingNodeIdxAfterMove, tokensAfterMove), getToken(movingNodeIdx, tokensAfterMove));
|
|
||||||
Arrays.sort(toFetchExpected);
|
|
||||||
|
|
||||||
assertEquals(Arrays.equals(toStream, toStreamExpected), true);
|
|
||||||
assertEquals(Arrays.equals(toFetch, toFetchExpected), true);
|
|
||||||
}
|
|
||||||
|
|
||||||
@SuppressWarnings("unchecked")
|
|
||||||
@Test
|
|
||||||
public void testMoveBeforePreviousNeighbor() throws UnknownHostException
|
|
||||||
{
|
|
||||||
// moves before its previous neighbor in the ring
|
|
||||||
|
|
||||||
int movingNodeIdx = 1;
|
|
||||||
int movingNodeIdxAfterMove = 7;
|
|
||||||
BigIntegerToken newToken = new BigIntegerToken("158873535527910577765226390751398592512");
|
|
||||||
BigIntegerToken[] tokens = initTokens();
|
|
||||||
BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
|
|
||||||
Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
|
|
||||||
|
|
||||||
Range<Token>[] toStream = ranges.left.toArray(new Range[0]);
|
|
||||||
Range<Token>[] toFetch = ranges.right.toArray(new Range[0]);
|
|
||||||
Arrays.sort(toStream);
|
|
||||||
Arrays.sort(toFetch);
|
|
||||||
|
|
||||||
Range<Token>[] toStreamExpected = new Range[2];
|
|
||||||
toStreamExpected[0] = new Range<Token>(getToken(movingNodeIdx, tokensAfterMove), getToken(movingNodeIdx - 1, tokensAfterMove));
|
|
||||||
toStreamExpected[1] = new Range<Token>(getToken(movingNodeIdx - 1, tokens), getToken(movingNodeIdx, tokens));
|
|
||||||
Arrays.sort(toStreamExpected);
|
|
||||||
Range<Token>[] toFetchExpected = new Range[1];
|
|
||||||
toFetchExpected[0] = new Range<Token>(getToken(movingNodeIdxAfterMove - 1, tokens), getToken(movingNodeIdxAfterMove, tokens));
|
|
||||||
Arrays.sort(toFetchExpected);
|
|
||||||
|
|
||||||
System.out.println("toStream : " + Arrays.toString(toStream));
|
|
||||||
System.out.println("toFetch : " + Arrays.toString(toFetch));
|
|
||||||
System.out.println("toStreamExpected : " + Arrays.toString(toStreamExpected));
|
|
||||||
System.out.println("toFetchExpected : " + Arrays.toString(toFetchExpected));
|
|
||||||
|
|
||||||
assertEquals(Arrays.equals(toStream, toStreamExpected), true);
|
|
||||||
assertEquals(Arrays.equals(toFetch, toFetchExpected), true);
|
|
||||||
}
|
|
||||||
|
|
||||||
private BigIntegerToken[] initTokensAfterMove(BigIntegerToken[] tokens,
|
|
||||||
int movingNodeIdx, BigIntegerToken newToken)
|
|
||||||
{
|
|
||||||
BigIntegerToken[] tokensAfterMove = tokens.clone();
|
|
||||||
tokensAfterMove[movingNodeIdx] = newToken;
|
|
||||||
return tokensAfterMove;
|
|
||||||
}
|
|
||||||
|
|
||||||
private BigIntegerToken[] initTokens()
|
|
||||||
{
|
|
||||||
BigIntegerToken[] tokens = new BigIntegerToken[] {
|
|
||||||
new BigIntegerToken("0"), // just to be able to test
|
|
||||||
new BigIntegerToken("34028236692093846346337460743176821145"),
|
|
||||||
new BigIntegerToken("42535295865117307932921825928971026432"),
|
|
||||||
new BigIntegerToken("63802943797675961899382738893456539648"),
|
|
||||||
new BigIntegerToken("85070591730234615865843651857942052864"),
|
|
||||||
new BigIntegerToken("106338239662793269832304564822427566080"),
|
|
||||||
new BigIntegerToken("127605887595351923798765477786913079296"),
|
|
||||||
new BigIntegerToken("148873535527910577765226390751398592512")
|
|
||||||
};
|
|
||||||
return tokens;
|
|
||||||
}
|
|
||||||
|
|
||||||
private TokenMetadata initTokenMetadata(BigIntegerToken[] tokens)
|
|
||||||
throws UnknownHostException
|
|
||||||
{
|
|
||||||
TokenMetadata tokenMetadataCurrent = new TokenMetadata();
|
|
||||||
|
|
||||||
int lastIPPart = 1;
|
|
||||||
for (BigIntegerToken token : tokens)
|
|
||||||
tokenMetadataCurrent.updateNormalToken(token, InetAddressAndPort.getByName("254.0.0." + Integer.toString(lastIPPart++)));
|
|
||||||
|
|
||||||
return tokenMetadataCurrent;
|
|
||||||
}
|
|
||||||
|
|
||||||
private BigIntegerToken getToken(int idx, BigIntegerToken[] tokens)
|
|
||||||
{
|
|
||||||
if (idx >= tokens.length)
|
|
||||||
idx = idx % tokens.length;
|
|
||||||
while (idx < 0)
|
|
||||||
idx += tokens.length;
|
|
||||||
|
|
||||||
return tokens[idx];
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
private Pair<Set<Range<Token>>, Set<Range<Token>>> calculateStreamAndFetchRanges(BigIntegerToken[] tokens, BigIntegerToken[] tokensAfterMove, int movingNodeIdx) throws UnknownHostException
|
|
||||||
{
|
|
||||||
RackInferringSnitch endpointSnitch = new RackInferringSnitch();
|
|
||||||
|
|
||||||
InetAddressAndPort movingNode = InetAddressAndPort.getByName("254.0.0." + Integer.toString(movingNodeIdx + 1));
|
|
||||||
|
|
||||||
|
|
||||||
TokenMetadata tokenMetadataCurrent = initTokenMetadata(tokens);
|
|
||||||
TokenMetadata tokenMetadataAfterMove = initTokenMetadata(tokensAfterMove);
|
|
||||||
AbstractReplicationStrategy strategy = new OldNetworkTopologyStrategy("Keyspace1", tokenMetadataCurrent, endpointSnitch, optsWithRF(2));
|
|
||||||
|
|
||||||
RangesAtEndpoint currentRanges = strategy.getAddressReplicas().get(movingNode);
|
|
||||||
RangesAtEndpoint updatedRanges = strategy.getPendingAddressRanges(tokenMetadataAfterMove, tokensAfterMove[movingNodeIdx], movingNode);
|
|
||||||
|
|
||||||
return asRanges(RangeRelocator.calculateStreamAndFetchRanges(currentRanges, updatedRanges));
|
|
||||||
}
|
|
||||||
|
|
||||||
private static Map<String, String> optsWithRF(int rf)
|
|
||||||
{
|
|
||||||
return Collections.singletonMap("replication_factor", Integer.toString(rf));
|
|
||||||
}
|
|
||||||
|
|
||||||
public static Pair<Set<Range<Token>>, Set<Range<Token>>> asRanges(Pair<RangesAtEndpoint, RangesAtEndpoint> replicas)
|
|
||||||
{
|
|
||||||
Set<Range<Token>> leftRanges = replicas.left.ranges();
|
|
||||||
Set<Range<Token>> rightRanges = replicas.right.ranges();
|
|
||||||
return Pair.create(leftRanges, rightRanges);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -67,7 +67,6 @@ public class ReplicationStrategyEndpointCacheTest
|
||||||
public void testEndpointsWereCached() throws Exception
|
public void testEndpointsWereCached() throws Exception
|
||||||
{
|
{
|
||||||
runEndpointsWereCachedTest(FakeSimpleStrategy.class, null);
|
runEndpointsWereCachedTest(FakeSimpleStrategy.class, null);
|
||||||
runEndpointsWereCachedTest(FakeOldNetworkTopologyStrategy.class, null);
|
|
||||||
runEndpointsWereCachedTest(FakeNetworkTopologyStrategy.class, new HashMap<String, String>());
|
runEndpointsWereCachedTest(FakeNetworkTopologyStrategy.class, new HashMap<String, String>());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -81,7 +80,6 @@ public class ReplicationStrategyEndpointCacheTest
|
||||||
public void testCacheRespectsTokenChanges() throws Exception
|
public void testCacheRespectsTokenChanges() throws Exception
|
||||||
{
|
{
|
||||||
runCacheRespectsTokenChangesTest(SimpleStrategy.class, null);
|
runCacheRespectsTokenChangesTest(SimpleStrategy.class, null);
|
||||||
runCacheRespectsTokenChangesTest(OldNetworkTopologyStrategy.class, null);
|
|
||||||
runCacheRespectsTokenChangesTest(NetworkTopologyStrategy.class, new HashMap<String, String>());
|
runCacheRespectsTokenChangesTest(NetworkTopologyStrategy.class, new HashMap<String, String>());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -135,23 +133,6 @@ public class ReplicationStrategyEndpointCacheTest
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
protected static class FakeOldNetworkTopologyStrategy extends OldNetworkTopologyStrategy
|
|
||||||
{
|
|
||||||
private boolean called = false;
|
|
||||||
|
|
||||||
public FakeOldNetworkTopologyStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
|
|
||||||
{
|
|
||||||
super(keyspaceName, tokenMetadata, snitch, configOptions);
|
|
||||||
}
|
|
||||||
|
|
||||||
public EndpointsForRange calculateNaturalReplicas(Token token, TokenMetadata metadata)
|
|
||||||
{
|
|
||||||
assert !called : "calculateNaturalReplicas was already called, result should have been cached";
|
|
||||||
called = true;
|
|
||||||
return super.calculateNaturalReplicas(token, metadata);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
protected static class FakeNetworkTopologyStrategy extends NetworkTopologyStrategy
|
protected static class FakeNetworkTopologyStrategy extends NetworkTopologyStrategy
|
||||||
{
|
{
|
||||||
private boolean called = false;
|
private boolean called = false;
|
||||||
|
|
|
||||||
|
|
@ -47,7 +47,7 @@ import org.apache.cassandra.db.marshal.UTF8Type;
|
||||||
import org.apache.cassandra.exceptions.ConfigurationException;
|
import org.apache.cassandra.exceptions.ConfigurationException;
|
||||||
import org.apache.cassandra.io.sstable.Component;
|
import org.apache.cassandra.io.sstable.Component;
|
||||||
import org.apache.cassandra.io.sstable.Descriptor;
|
import org.apache.cassandra.io.sstable.Descriptor;
|
||||||
import org.apache.cassandra.locator.OldNetworkTopologyStrategy;
|
import org.apache.cassandra.locator.NetworkTopologyStrategy;
|
||||||
import org.apache.cassandra.utils.FBUtilities;
|
import org.apache.cassandra.utils.FBUtilities;
|
||||||
|
|
||||||
import static org.apache.cassandra.Util.throwAssert;
|
import static org.apache.cassandra.Util.throwAssert;
|
||||||
|
|
@ -385,7 +385,7 @@ public class MigrationManagerTest
|
||||||
}
|
}
|
||||||
|
|
||||||
Map<String, String> replicationMap = new HashMap<>();
|
Map<String, String> replicationMap = new HashMap<>();
|
||||||
replicationMap.put(ReplicationParams.CLASS, OldNetworkTopologyStrategy.class.getName());
|
replicationMap.put(ReplicationParams.CLASS, NetworkTopologyStrategy.class.getName());
|
||||||
replicationMap.put("replication_factor", "1");
|
replicationMap.put("replication_factor", "1");
|
||||||
|
|
||||||
KeyspaceMetadata newKs = KeyspaceMetadata.create(cf.keyspace, KeyspaceParams.create(true, replicationMap));
|
KeyspaceMetadata newKs = KeyspaceMetadata.create(cf.keyspace, KeyspaceParams.create(true, replicationMap));
|
||||||
|
|
|
||||||
|
|
@ -190,7 +190,7 @@ public class BootstrapTransientTest
|
||||||
true,
|
true,
|
||||||
tmds.left,
|
tmds.left,
|
||||||
tmds.right,
|
tmds.right,
|
||||||
"OldNetworkTopologyStrategyTest",
|
"TestKeyspace",
|
||||||
sourceFilters);
|
sourceFilters);
|
||||||
result.asMap().forEach((replica, list) -> System.out.printf("Replica %s, sources %s%n", replica, list));
|
result.asMap().forEach((replica, list) -> System.out.printf("Replica %s, sources %s%n", replica, list));
|
||||||
assertMultimapEqualsIgnoreOrder(expectedResult, result);
|
assertMultimapEqualsIgnoreOrder(expectedResult, result);
|
||||||
|
|
|
||||||
|
|
@ -589,7 +589,7 @@ public class MoveTransientTest
|
||||||
true,
|
true,
|
||||||
tmds.left,
|
tmds.left,
|
||||||
tmds.right,
|
tmds.right,
|
||||||
"OldNetworkTopologyStrategyTest",
|
"TestKeyspace",
|
||||||
sourceFilters);
|
sourceFilters);
|
||||||
logger.info("Ranges to fetch with preferred endpoints");
|
logger.info("Ranges to fetch with preferred endpoints");
|
||||||
logger.info(result.toString());
|
logger.info(result.toString());
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue