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

configurable options for replication strategies 

Patch by Jeremy Hanna; review by eevans for CASSANDRA-1066

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@981069 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Eric Evans 2010-07-31 15:42:27 +00:00
parent 170bd4d0a0
commit df8a93358c
35 changed files with 427 additions and 302 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
conf/cassandra.yaml
View File

@ -185,7 +185,8 @@ request_scheduler_id: keyspace
# - name: name of the keyspace; "system" and "definitions" are # - name: name of the keyspace; "system" and "definitions" are
# reserved for Cassandra Internals. # reserved for Cassandra Internals.
# - replica_placement_strategy: the class that determines how replicas # - replica_placement_strategy: the class that determines how replicas
# are distributed among nodes. Must extend AbstractReplicationStrategy. # are distributed among nodes. Contains both the class as well as
# configuration information. Must extend AbstractReplicationStrategy.
# Out of the box, Cassandra provides # Out of the box, Cassandra provides
# * org.apache.cassandra.locator.RackUnawareStrategy # * org.apache.cassandra.locator.RackUnawareStrategy
# * org.apache.cassandra.locator.RackAwareStrategy # * org.apache.cassandra.locator.RackAwareStrategy
@ -201,9 +202,17 @@ request_scheduler_id: keyspace
# different rack in in the first. # different rack in in the first.
# #
# DatacenterShardStrategy is a generalization of RackAwareStrategy. # DatacenterShardStrategy is a generalization of RackAwareStrategy.
# For each datacenter, you can specify (in `datacenter.properties`) # For each datacenter, you can specify how many replicas you want
# how many replicas you want on a per-keyspace basis. Replicas are # on a per-keyspace basis. Replicas are placed on different racks
# placed on different racks within each DC, if possible. # within each DC, if possible. This strategy also requires rack aware
# snitch, such as RackInferringSnitch or PropertyFileSnitch.
# An example:
# - name: Keyspace1
# replica_placement_strategy: org.apache.cassandra.locator.DatacenterShardStrategy
# strategy_options:
# DC1 : 3
# DC2 : 2
# DC3 : 1
# #
# - replication_factor: Number of replicas of each row # - replication_factor: Number of replicas of each row
# - column_families: column families associated with this keyspace # - column_families: column families associated with this keyspace
@ -236,8 +245,8 @@ request_scheduler_id: keyspace
# and 1. defaults to 1.0 (always read repair). # and 1. defaults to 1.0 (always read repair).
# - preload_row_cache: If true, will populate row cache on startup. # - preload_row_cache: If true, will populate row cache on startup.
# Defaults to false. # Defaults to false.
# - gc_grace_seconds: specifies the time to wait before garbage # - gc_grace_seconds: specifies the time to wait before garbage
# collecting tombstones (deletion markers). defaults to 864000 (10 # collecting tombstones (deletion markers). defaults to 864000 (10
# days). See http://wiki.apache.org/cassandra/DistributedDeletes # days). See http://wiki.apache.org/cassandra/DistributedDeletes
# #
# NOTE: this keyspace definition is for demonstration purposes only. # NOTE: this keyspace definition is for demonstration purposes only.

1
interface/cassandra.genavro
View File

@ -120,6 +120,7 @@ protocol Cassandra {
record KsDef { record KsDef {
string name; string name;
string strategy_class; string strategy_class;
union{ map<string>, null } strategy_options;
int replication_factor; int replication_factor;
array<CfDef> cf_defs; array<CfDef> cf_defs;
} }

5
interface/cassandra.thrift
View File

@ -375,8 +375,9 @@ struct CfDef {
struct KsDef { struct KsDef {
1: required string name, 1: required string name,
2: required string strategy_class, 2: required string strategy_class,
3: required i32 replication_factor, 3: optional map<string,string> strategy_options,
5: required list<CfDef> cf_defs, 4: required i32 replication_factor,
5: required list<CfDef> cf_defs,
} }
service Cassandra { service Cassandra {

15
src/java/org/apache/cassandra/avro/CassandraServer.java
View File

@ -661,10 +661,23 @@ public class CassandraServer implements Cassandra {
Collections.<byte[], ColumnDefinition>emptyMap()); Collections.<byte[], ColumnDefinition>emptyMap());
cfDefs.add(cfmeta); cfDefs.add(cfmeta);
} }
Map<String, String> strategyOptions = null;
if (ksDef.strategy_options != null && !ksDef.strategy_options.isEmpty())
{
strategyOptions = new HashMap<String, String>();
for (Map.Entry<Utf8, Utf8> option : ksDef.strategy_options.entrySet())
{
strategyOptions.put(option.getKey().toString(), option.getValue().toString());
}
}
KSMetaData ksmeta = new KSMetaData( KSMetaData ksmeta = new KSMetaData(
ksDef.name.toString(), ksDef.name.toString(),
(Class<? extends AbstractReplicationStrategy>)Class.forName(ksDef.strategy_class.toString()), (Class<? extends AbstractReplicationStrategy>)Class.forName(ksDef.strategy_class.toString()),
strategyOptions,
(int)ksDef.replication_factor, (int)ksDef.replication_factor,
cfDefs.toArray(new CFMetaData[cfDefs.size()])); cfDefs.toArray(new CFMetaData[cfDefs.size()]));
AddKeyspace add = new AddKeyspace(ksmeta); AddKeyspace add = new AddKeyspace(ksmeta);

4
src/java/org/apache/cassandra/client/RingCache.java
View File

@ -118,7 +118,7 @@ public class RingCache
{ {
if (tokenMetadata == null) if (tokenMetadata == null)
throw new RuntimeException("Must refresh endpoints before looking up a key."); throw new RuntimeException("Must refresh endpoints before looking up a key.");
AbstractReplicationStrategy strat = StorageService.getReplicationStrategy(tokenMetadata, keyspace); AbstractReplicationStrategy strat = StorageService.createReplicationStrategy(tokenMetadata, keyspace);
return strat.getNaturalEndpoints(partitioner_.getToken(key), keyspace); return strat.getNaturalEndpoints(partitioner_.getToken(key));
} }
} }

27
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -344,11 +344,15 @@ public class DatabaseDescriptor
CommitLog.setSegmentSize(conf.commitlog_rotation_threshold_in_mb * 1024 * 1024); CommitLog.setSegmentSize(conf.commitlog_rotation_threshold_in_mb * 1024 * 1024);
// Hardcoded system tables // Hardcoded system tables
KSMetaData systemMeta = new KSMetaData(Table.SYSTEM_TABLE, LocalStrategy.class, 1, new CFMetaData[]{CFMetaData.StatusCf, KSMetaData systemMeta = new KSMetaData(Table.SYSTEM_TABLE,
CFMetaData.HintsCf, LocalStrategy.class,
CFMetaData.MigrationsCf, null,
CFMetaData.SchemaCf, 1,
CFMetaData.StatisticsCf new CFMetaData[]{CFMetaData.StatusCf,
CFMetaData.HintsCf,
CFMetaData.MigrationsCf,
CFMetaData.SchemaCf,
CFMetaData.StatisticsCf
}); });
CFMetaData.map(CFMetaData.StatusCf); CFMetaData.map(CFMetaData.StatusCf);
CFMetaData.map(CFMetaData.HintsCf); CFMetaData.map(CFMetaData.HintsCf);
@ -618,8 +622,11 @@ public class DatabaseDescriptor
cf.gc_grace_seconds, cf.gc_grace_seconds,
metadata); metadata);
} }
defs.add(new KSMetaData(keyspace.name, strategyClass, keyspace.replication_factor, cfDefs)); defs.add(new KSMetaData(keyspace.name,
strategyClass,
keyspace.strategy_options,
keyspace.replication_factor,
cfDefs));
} }
return defs; return defs;
@ -750,6 +757,12 @@ public class DatabaseDescriptor
throw new RuntimeException(table + " not found. Failure to call loadSchemas() perhaps?"); throw new RuntimeException(table + " not found. Failure to call loadSchemas() perhaps?");
return meta.strategyClass; return meta.strategyClass;
} }
public static KSMetaData getKSMetaData(String table)
{
assert table != null;
return tables.get(table);
}
public static String getJobTrackerAddress() public static String getJobTrackerAddress()
{ {

26
src/java/org/apache/cassandra/config/KSMetaData.java
View File

@ -36,13 +36,15 @@ public final class KSMetaData
{ {
public final String name; public final String name;
public final Class<? extends AbstractReplicationStrategy> strategyClass; public final Class<? extends AbstractReplicationStrategy> strategyClass;
public final Map<String, String> strategyOptions;
public final int replicationFactor; public final int replicationFactor;
private final Map<String, CFMetaData> cfMetaData; private final Map<String, CFMetaData> cfMetaData;
public KSMetaData(String name, Class<? extends AbstractReplicationStrategy> strategyClass, int replicationFactor, CFMetaData... cfDefs) public KSMetaData(String name, Class<? extends AbstractReplicationStrategy> strategyClass, Map<String, String> strategyOptions, int replicationFactor, CFMetaData... cfDefs)
{ {
this.name = name; this.name = name;
this.strategyClass = strategyClass == null ? RackUnawareStrategy.class : strategyClass; this.strategyClass = strategyClass == null ? RackUnawareStrategy.class : strategyClass;
this.strategyOptions = strategyOptions;
this.replicationFactor = replicationFactor; this.replicationFactor = replicationFactor;
Map<String, CFMetaData> cfmap = new HashMap<String, CFMetaData>(); Map<String, CFMetaData> cfmap = new HashMap<String, CFMetaData>();
for (CFMetaData cfm : cfDefs) for (CFMetaData cfm : cfDefs)
@ -59,6 +61,7 @@ public final class KSMetaData
KSMetaData other = (KSMetaData)obj; KSMetaData other = (KSMetaData)obj;
return other.name.equals(name) return other.name.equals(name)
&& ObjectUtils.equals(other.strategyClass, strategyClass) && ObjectUtils.equals(other.strategyClass, strategyClass)
&& ObjectUtils.equals(other.strategyOptions, strategyOptions)
&& other.replicationFactor == replicationFactor && other.replicationFactor == replicationFactor
&& other.cfMetaData.size() == cfMetaData.size() && other.cfMetaData.size() == cfMetaData.size()
&& other.cfMetaData.equals(cfMetaData); && other.cfMetaData.equals(cfMetaData);
@ -74,6 +77,14 @@ public final class KSMetaData
org.apache.cassandra.avro.KsDef ks = new org.apache.cassandra.avro.KsDef(); org.apache.cassandra.avro.KsDef ks = new org.apache.cassandra.avro.KsDef();
ks.name = new Utf8(name); ks.name = new Utf8(name);
ks.strategy_class = new Utf8(strategyClass.getName()); ks.strategy_class = new Utf8(strategyClass.getName());
if (strategyOptions != null)
{
ks.strategy_options = new HashMap<Utf8, Utf8>();
for (Map.Entry<String, String> e : strategyOptions.entrySet())
{
ks.strategy_options.put(new Utf8(e.getKey()), new Utf8(e.getValue()));
}
}
ks.replication_factor = replicationFactor; ks.replication_factor = replicationFactor;
ks.cf_defs = SerDeUtils.createArray(cfMetaData.size(), org.apache.cassandra.avro.CfDef.SCHEMA$); ks.cf_defs = SerDeUtils.createArray(cfMetaData.size(), org.apache.cassandra.avro.CfDef.SCHEMA$);
for (CFMetaData cfm : cfMetaData.values()) for (CFMetaData cfm : cfMetaData.values())
@ -83,7 +94,7 @@ public final class KSMetaData
public static KSMetaData inflate(org.apache.cassandra.avro.KsDef ks) throws ConfigurationException public static KSMetaData inflate(org.apache.cassandra.avro.KsDef ks) throws ConfigurationException
{ {
Class<AbstractReplicationStrategy> repStratClass = null; Class<AbstractReplicationStrategy> repStratClass;
try try
{ {
repStratClass = (Class<AbstractReplicationStrategy>)Class.forName(ks.strategy_class.toString()); repStratClass = (Class<AbstractReplicationStrategy>)Class.forName(ks.strategy_class.toString());
@ -92,12 +103,21 @@ public final class KSMetaData
{ {
throw new ConfigurationException("Could not create ReplicationStrategy of type " + ks.strategy_class, ex); throw new ConfigurationException("Could not create ReplicationStrategy of type " + ks.strategy_class, ex);
} }
Map<String, String> strategyOptions = null;
if (ks.strategy_options != null)
{
strategyOptions = new HashMap<String, String>();
for (Map.Entry<Utf8, Utf8> e : ks.strategy_options.entrySet())
{
strategyOptions.put(e.getKey().toString(), e.getValue().toString());
}
}
int cfsz = (int)ks.cf_defs.size(); int cfsz = (int)ks.cf_defs.size();
CFMetaData[] cfMetaData = new CFMetaData[cfsz]; CFMetaData[] cfMetaData = new CFMetaData[cfsz];
Iterator<org.apache.cassandra.avro.CfDef> cfiter = ks.cf_defs.iterator(); Iterator<org.apache.cassandra.avro.CfDef> cfiter = ks.cf_defs.iterator();
for (int i = 0; i < cfsz; i++) for (int i = 0; i < cfsz; i++)
cfMetaData[i] = CFMetaData.inflate(cfiter.next()); cfMetaData[i] = CFMetaData.inflate(cfiter.next());
return new KSMetaData(ks.name.toString(), repStratClass, ks.replication_factor, cfMetaData); return new KSMetaData(ks.name.toString(), repStratClass, strategyOptions, ks.replication_factor, cfMetaData);
} }
} }

3
src/java/org/apache/cassandra/config/RawKeyspace.java
View File

@ -1,5 +1,7 @@
package org.apache.cassandra.config; package org.apache.cassandra.config;
import java.util.Map;
/** /**
* @deprecated Yaml configuration for Keyspaces and ColumnFamilies is deprecated in 0.7 * @deprecated Yaml configuration for Keyspaces and ColumnFamilies is deprecated in 0.7
*/ */
@ -7,6 +9,7 @@ public class RawKeyspace
{ {
public String name; public String name;
public String replica_placement_strategy; public String replica_placement_strategy;
public Map<String,String> strategy_options;
public Integer replication_factor; public Integer replication_factor;
public RawColumnFamily[] column_families; public RawColumnFamily[] column_families;
} }

9
src/java/org/apache/cassandra/config/ReplicationStrategy.java Normal file
View File

@ -0,0 +1,9 @@
package org.apache.cassandra.config;
import java.util.Map;
public class ReplicationStrategy
{
public String strategy_class;
public Map<String, String> strategy_options;
}

2
src/java/org/apache/cassandra/db/migration/AddColumnFamily.java
View File

@ -82,7 +82,7 @@ public class AddColumnFamily extends Migration
{ {
List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values()); List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values());
newCfs.add(cfm); newCfs.add(cfm);
return new KSMetaData(ksm.name, ksm.strategyClass, ksm.replicationFactor, newCfs.toArray(new CFMetaData[newCfs.size()])); return new KSMetaData(ksm.name, ksm.strategyClass, ksm.strategyOptions, ksm.replicationFactor, newCfs.toArray(new CFMetaData[newCfs.size()]));
} }
public void applyModels() throws IOException public void applyModels() throws IOException

2
src/java/org/apache/cassandra/db/migration/DropColumnFamily.java
View File

@ -81,7 +81,7 @@ public class DropColumnFamily extends Migration
List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values()); List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values());
newCfs.remove(cfm); newCfs.remove(cfm);
assert newCfs.size() == ksm.cfMetaData().size() - 1; assert newCfs.size() == ksm.cfMetaData().size() - 1;
return new KSMetaData(ksm.name, ksm.strategyClass, ksm.replicationFactor, newCfs.toArray(new CFMetaData[newCfs.size()])); return new KSMetaData(ksm.name, ksm.strategyClass, ksm.strategyOptions, ksm.replicationFactor, newCfs.toArray(new CFMetaData[newCfs.size()]));
} }
@Override @Override

2
src/java/org/apache/cassandra/db/migration/RenameColumnFamily.java
View File

@ -91,7 +91,7 @@ public class RenameColumnFamily extends Migration
assert newCfs.size() == ksm.cfMetaData().size() - 1; assert newCfs.size() == ksm.cfMetaData().size() - 1;
CFMetaData newCfm = CFMetaData.rename(oldCfm, newName); CFMetaData newCfm = CFMetaData.rename(oldCfm, newName);
newCfs.add(newCfm); newCfs.add(newCfm);
return new KSMetaData(ksm.name, ksm.strategyClass, ksm.replicationFactor, newCfs.toArray(new CFMetaData[newCfs.size()])); return new KSMetaData(ksm.name, ksm.strategyClass, ksm.strategyOptions, ksm.replicationFactor, newCfs.toArray(new CFMetaData[newCfs.size()]));
} }
@Override @Override

2
src/java/org/apache/cassandra/db/migration/RenameKeyspace.java
View File

@ -85,7 +85,7 @@ public class RenameKeyspace extends Migration
CFMetaData.purge(oldCf); CFMetaData.purge(oldCf);
newCfs.add(CFMetaData.renameTable(oldCf, newName)); newCfs.add(CFMetaData.renameTable(oldCf, newName));
} }
return new KSMetaData(newName, ksm.strategyClass, ksm.replicationFactor, newCfs.toArray(new CFMetaData[newCfs.size()])); return new KSMetaData(newName, ksm.strategyClass, ksm.strategyOptions, ksm.replicationFactor, newCfs.toArray(new CFMetaData[newCfs.size()]));
} }
@Override @Override

4
src/java/org/apache/cassandra/dht/BootStrapper.java
View File

@ -149,10 +149,10 @@ public class BootStrapper
{ {
assert tokenMetadata.sortedTokens().size() > 0; assert tokenMetadata.sortedTokens().size() > 0;
final AbstractReplicationStrategy strat = StorageService.instance.getReplicationStrategy(table); final AbstractReplicationStrategy strat = StorageService.instance.getReplicationStrategy(table);
Collection<Range> myRanges = strat.getPendingAddressRanges(tokenMetadata, token, address, table); Collection<Range> myRanges = strat.getPendingAddressRanges(tokenMetadata, token, address);
Multimap<Range, InetAddress> myRangeAddresses = ArrayListMultimap.create(); Multimap<Range, InetAddress> myRangeAddresses = ArrayListMultimap.create();
Multimap<Range, InetAddress> rangeAddresses = strat.getRangeAddresses(tokenMetadata, table); Multimap<Range, InetAddress> rangeAddresses = strat.getRangeAddresses(tokenMetadata);
for (Range myRange : myRanges) for (Range myRange : myRanges)
{ {
for (Range range : rangeAddresses.keySet()) for (Range range : rangeAddresses.keySet())

131
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
View File

@ -19,12 +19,16 @@
package org.apache.cassandra.locator; package org.apache.cassandra.locator;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.net.InetAddress; import java.net.InetAddress;
import java.util.ArrayList; import java.util.*;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.service.*;
import org.apache.commons.lang.ObjectUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -34,13 +38,8 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token; import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.FailureDetector; import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.service.AbstractWriteResponseHandler;
import org.apache.cassandra.service.IResponseResolver;
import org.apache.cassandra.service.QuorumResponseHandler;
import org.apache.cassandra.service.WriteResponseHandler;
import org.apache.cassandra.thrift.ConsistencyLevel; import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.cliffc.high_scale_lib.NonBlockingHashMap; import org.cliffc.high_scale_lib.NonBlockingHashMap;
/** /**
@ -50,42 +49,44 @@ public abstract class AbstractReplicationStrategy
{ {
private static final Logger logger = LoggerFactory.getLogger(AbstractReplicationStrategy.class); private static final Logger logger = LoggerFactory.getLogger(AbstractReplicationStrategy.class);
public String table;
private TokenMetadata tokenMetadata; private TokenMetadata tokenMetadata;
protected final IEndpointSnitch snitch; public final IEndpointSnitch snitch;
private volatile Map<EndpointCacheKey, ArrayList<InetAddress>> cachedEndpoints; private volatile Map<Token, ArrayList<InetAddress>> cachedEndpoints;
public Map<String, String> configOptions;
AbstractReplicationStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) AbstractReplicationStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
{ {
assert table != null;
assert snitch != null; assert snitch != null;
assert tokenMetadata != null; assert tokenMetadata != null;
this.tokenMetadata = tokenMetadata; this.tokenMetadata = tokenMetadata;
this.snitch = snitch; this.snitch = snitch;
cachedEndpoints = new NonBlockingHashMap<EndpointCacheKey, ArrayList<InetAddress>>(); cachedEndpoints = new NonBlockingHashMap<Token, ArrayList<InetAddress>>();
this.tokenMetadata.register(this); this.tokenMetadata.register(this);
this.configOptions = configOptions;
this.table = table;
} }
/** /**
* get the (possibly cached) endpoints that should store the given Token, for the given table. * get the (possibly cached) endpoints that should store the given Token
* Note that while the endpoints are conceptually a Set (no duplicates will be included), * Note that while the endpoints are conceptually a Set (no duplicates will be included),
* we return a List to avoid an extra allocation when sorting by proximity later * we return a List to avoid an extra allocation when sorting by proximity later
* @param searchToken the token the natural endpoints are requested for * @param searchToken the token the natural endpoints are requested for
* @param table the table the natural endpoints are requested for * @return a copy of the natural endpoints for the given token
* @return a copy of the natural endpoints for the given token and table * @throws IllegalStateException if the number of requested replicas is greater than the number of known endpoints
* @throws IllegalStateException if the number of requested replicas is greater than the number of known endpints
*/ */
public ArrayList<InetAddress> getNaturalEndpoints(Token searchToken, String table) throws IllegalStateException public ArrayList<InetAddress> getNaturalEndpoints(Token searchToken) throws IllegalStateException
{ {
int replicas = getReplicationFactor(table); int replicas = getReplicationFactor();
Token keyToken = TokenMetadata.firstToken(tokenMetadata.sortedTokens(), searchToken); Token keyToken = TokenMetadata.firstToken(tokenMetadata.sortedTokens(), searchToken);
EndpointCacheKey cacheKey = new EndpointCacheKey(table, keyToken); ArrayList<InetAddress> endpoints = cachedEndpoints.get(keyToken);
ArrayList<InetAddress> endpoints = cachedEndpoints.get(cacheKey);
if (endpoints == null) if (endpoints == null)
{ {
TokenMetadata tokenMetadataClone = tokenMetadata.cloneOnlyTokenMap(); TokenMetadata tokenMetadataClone = tokenMetadata.cloneOnlyTokenMap();
keyToken = TokenMetadata.firstToken(tokenMetadataClone.sortedTokens(), searchToken); keyToken = TokenMetadata.firstToken(tokenMetadataClone.sortedTokens(), searchToken);
cacheKey = new EndpointCacheKey(table, keyToken); endpoints = new ArrayList<InetAddress>(calculateNaturalEndpoints(searchToken, tokenMetadataClone));
endpoints = new ArrayList<InetAddress>(calculateNaturalEndpoints(searchToken, tokenMetadataClone, table)); cachedEndpoints.put(keyToken, endpoints);
cachedEndpoints.put(cacheKey, endpoints);
} }
// calculateNaturalEndpoints should have checked this already, this is a safety // calculateNaturalEndpoints should have checked this already, this is a safety
@ -95,27 +96,25 @@ public abstract class AbstractReplicationStrategy
} }
/** /**
* calculate the natural endpionts for the given token, for the given table. * calculate the natural endpoints for the given token
* *
* @see #getNaturalEndpoints(org.apache.cassandra.dht.Token, String) * @see #getNaturalEndpoints(org.apache.cassandra.dht.Token)
* *
* @param searchToken the token the natural endpoints are requested for * @param searchToken the token the natural endpoints are requested for
* @param table the table the natural endpoints are requested for * @return a copy of the natural endpoints for the given token
* @return a copy of the natural endpoints for the given token and table
* @throws IllegalStateException if the number of requested replicas is greater than the number of known endpoints * @throws IllegalStateException if the number of requested replicas is greater than the number of known endpoints
*/ */
public abstract Set<InetAddress> calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata, String table) throws IllegalStateException; public abstract Set<InetAddress> calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata) throws IllegalStateException;
public AbstractWriteResponseHandler getWriteResponseHandler(Collection<InetAddress> writeEndpoints, public AbstractWriteResponseHandler getWriteResponseHandler(Collection<InetAddress> writeEndpoints,
Multimap<InetAddress, InetAddress> hintedEndpoints, Multimap<InetAddress, InetAddress> hintedEndpoints,
ConsistencyLevel consistencyLevel, ConsistencyLevel consistencyLevel)
String table)
{ {
return new WriteResponseHandler(writeEndpoints, hintedEndpoints, consistencyLevel, table); return new WriteResponseHandler(writeEndpoints, hintedEndpoints, consistencyLevel, table);
} }
// instance method so test subclasses can override it // instance method so test subclasses can override it
int getReplicationFactor(String table) int getReplicationFactor()
{ {
return DatabaseDescriptor.getReplicationFactor(table); return DatabaseDescriptor.getReplicationFactor(table);
} }
@ -170,14 +169,14 @@ public abstract class AbstractReplicationStrategy
* (fixing this would probably require merging tokenmetadata into replicationstrategy, * (fixing this would probably require merging tokenmetadata into replicationstrategy,
* so we could cache/invalidate cleanly.) * so we could cache/invalidate cleanly.)
*/ */
public Multimap<InetAddress, Range> getAddressRanges(TokenMetadata metadata, String table) public Multimap<InetAddress, Range> getAddressRanges(TokenMetadata metadata)
{ {
Multimap<InetAddress, Range> map = HashMultimap.create(); Multimap<InetAddress, Range> map = HashMultimap.create();
for (Token token : metadata.sortedTokens()) for (Token token : metadata.sortedTokens())
{ {
Range range = metadata.getPrimaryRangeFor(token); Range range = metadata.getPrimaryRangeFor(token);
for (InetAddress ep : calculateNaturalEndpoints(token, metadata, table)) for (InetAddress ep : calculateNaturalEndpoints(token, metadata))
{ {
map.put(ep, range); map.put(ep, range);
} }
@ -186,14 +185,14 @@ public abstract class AbstractReplicationStrategy
return map; return map;
} }
public Multimap<Range, InetAddress> getRangeAddresses(TokenMetadata metadata, String table) public Multimap<Range, InetAddress> getRangeAddresses(TokenMetadata metadata)
{ {
Multimap<Range, InetAddress> map = HashMultimap.create(); Multimap<Range, InetAddress> map = HashMultimap.create();
for (Token token : metadata.sortedTokens()) for (Token token : metadata.sortedTokens())
{ {
Range range = metadata.getPrimaryRangeFor(token); Range range = metadata.getPrimaryRangeFor(token);
for (InetAddress ep : calculateNaturalEndpoints(token, metadata, table)) for (InetAddress ep : calculateNaturalEndpoints(token, metadata))
{ {
map.put(range, ep); map.put(range, ep);
} }
@ -202,32 +201,27 @@ public abstract class AbstractReplicationStrategy
return map; return map;
} }
public Multimap<InetAddress, Range> getAddressRanges(String table) public Multimap<InetAddress, Range> getAddressRanges()
{ {
return getAddressRanges(tokenMetadata, table); return getAddressRanges(tokenMetadata);
} }
public Collection<Range> getPendingAddressRanges(TokenMetadata metadata, Token pendingToken, InetAddress pendingAddress, String table) public Collection<Range> getPendingAddressRanges(TokenMetadata metadata, Token pendingToken, InetAddress pendingAddress)
{ {
TokenMetadata temp = metadata.cloneOnlyTokenMap(); TokenMetadata temp = metadata.cloneOnlyTokenMap();
temp.updateNormalToken(pendingToken, pendingAddress); temp.updateNormalToken(pendingToken, pendingAddress);
return getAddressRanges(temp, table).get(pendingAddress); return getAddressRanges(temp).get(pendingAddress);
} }
public QuorumResponseHandler getQuorumResponseHandler(IResponseResolver responseResolver, ConsistencyLevel consistencyLevel, String table) public QuorumResponseHandler getQuorumResponseHandler(IResponseResolver responseResolver, ConsistencyLevel consistencyLevel)
{ {
return new QuorumResponseHandler(responseResolver, consistencyLevel, table); return new QuorumResponseHandler(responseResolver, consistencyLevel, table);
} }
protected static class EndpointCacheKey extends Pair<String, Token>
{
public EndpointCacheKey(String table, Token keyToken) {super(table, keyToken);}
}
protected void clearCachedEndpoints() protected void clearCachedEndpoints()
{ {
logger.debug("clearing cached endpoints"); logger.debug("clearing cached endpoints");
cachedEndpoints = new NonBlockingHashMap<EndpointCacheKey, ArrayList<InetAddress>>(); cachedEndpoints = new NonBlockingHashMap<Token, ArrayList<InetAddress>>();
} }
public void invalidateCachedTokenEndpointValues() public void invalidateCachedTokenEndpointValues()
@ -239,4 +233,47 @@ public abstract class AbstractReplicationStrategy
{ {
clearCachedEndpoints(); clearCachedEndpoints();
} }
public static AbstractReplicationStrategy createReplicationStrategy(String table,
Class<? extends AbstractReplicationStrategy> strategyClass,
TokenMetadata tokenMetadata,
IEndpointSnitch snitch,
Map<String, String> strategyOptions)
throws ConfigurationException
{
AbstractReplicationStrategy strategy;
Class [] parameterTypes = new Class[] {String.class, TokenMetadata.class, IEndpointSnitch.class, Map.class};
try
{
Constructor<? extends AbstractReplicationStrategy> constructor = strategyClass.getConstructor(parameterTypes);
strategy = constructor.newInstance(table, tokenMetadata, snitch, strategyOptions);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
return strategy;
}
public static AbstractReplicationStrategy createReplicationStrategy(String table,
String strategyClassName,
TokenMetadata tokenMetadata,
IEndpointSnitch snitch,
Map<String, String> strategyOptions)
throws ConfigurationException
{
AbstractReplicationStrategy strategy;
Class<? extends AbstractReplicationStrategy> c;
try
{
c = (Class<? extends AbstractReplicationStrategy>) Class.forName(strategyClassName);
}
catch (ClassNotFoundException e)
{
throw new ConfigurationException("Invalid replication strategy class: " + strategyClassName);
}
return createReplicationStrategy(table, c, tokenMetadata, snitch, strategyOptions);
}
} }

65
src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java
View File

@ -21,10 +21,12 @@
package org.apache.cassandra.locator; package org.apache.cassandra.locator;
import java.io.*;
import java.net.InetAddress; import java.net.InetAddress;
import java.util.*; import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -33,8 +35,6 @@ import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.dht.Token; import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.*; import org.apache.cassandra.service.*;
import org.apache.cassandra.thrift.ConsistencyLevel; import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.utils.ResourceWatcher;
import org.apache.cassandra.utils.WrappedRunnable;
/** /**
* This Replication Strategy takes a property file that gives the intended * This Replication Strategy takes a property file that gives the intended
@ -51,51 +51,30 @@ import org.apache.cassandra.utils.WrappedRunnable;
*/ */
public class DatacenterShardStrategy extends AbstractReplicationStrategy public class DatacenterShardStrategy extends AbstractReplicationStrategy
{ {
private static final String DATACENTER_PROPERTY_FILENAME = "datacenters.properties";
private AbstractRackAwareSnitch snitch; private AbstractRackAwareSnitch snitch;
private volatile Map<String, Map<String, Integer>> datacenters; private Map<String, Integer> datacenters;
private static final Logger logger = LoggerFactory.getLogger(DatacenterShardStrategy.class); private static final Logger logger = LoggerFactory.getLogger(DatacenterShardStrategy.class);
public DatacenterShardStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) throws ConfigurationException public DatacenterShardStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions) throws ConfigurationException
{ {
super(tokenMetadata, snitch); super(table, tokenMetadata, snitch, configOptions);
if ((!(snitch instanceof AbstractRackAwareSnitch))) if ((!(snitch instanceof AbstractRackAwareSnitch)))
throw new IllegalArgumentException("DatacenterShardStrategy requires a rack-aware endpointsnitch"); throw new IllegalArgumentException("DatacenterShardStrategy requires a rack-aware endpointsnitch");
this.snitch = (AbstractRackAwareSnitch)snitch; this.snitch = (AbstractRackAwareSnitch)snitch;
reloadConfiguration();
Runnable runnable = new WrappedRunnable()
{
protected void runMayThrow() throws ConfigurationException
{
reloadConfiguration();
}
};
ResourceWatcher.watch(DATACENTER_PROPERTY_FILENAME, runnable, 60 * 1000);
}
public synchronized void reloadConfiguration() throws ConfigurationException Map<String, Integer> newDatacenters = new HashMap<String, Integer>();
{ for (Entry entry : configOptions.entrySet())
Properties props = PropertyFileSnitch.resourceToProperties(DATACENTER_PROPERTY_FILENAME);
Map<String, Map<String, Integer>> newDatacenters = new HashMap<String, Map<String, Integer>>();
for (Entry entry : props.entrySet())
{ {
String[] keys = ((String)entry.getKey()).split(":"); newDatacenters.put((String) entry.getKey(), Integer.parseInt((String) entry.getValue()));
Map<String, Integer> map = newDatacenters.get(keys[0]);
if (map == null)
map = new HashMap<String, Integer>();
map.put(keys[1], Integer.parseInt((String) entry.getValue()));
newDatacenters.put(keys[0], map);
} }
datacenters = Collections.unmodifiableMap(newDatacenters); datacenters = Collections.unmodifiableMap(newDatacenters);
logger.info(DATACENTER_PROPERTY_FILENAME + " changed, clearing endpoint cache");
clearCachedEndpoints();
} }
public Set<InetAddress> calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata, String table) public Set<InetAddress> calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata)
{ {
int totalReplicas = getReplicationFactor(table); int totalReplicas = getReplicationFactor();
Map<String, Integer> remainingReplicas = new HashMap<String, Integer>(datacenters.get(table)); Map<String, Integer> remainingReplicas = new HashMap<String, Integer>(datacenters);
Map<String, Set<String>> dcUsedRacks = new HashMap<String, Set<String>>(); Map<String, Set<String>> dcUsedRacks = new HashMap<String, Set<String>>();
Set<InetAddress> endpoints = new LinkedHashSet<InetAddress>(totalReplicas); Set<InetAddress> endpoints = new LinkedHashSet<InetAddress>(totalReplicas);
@ -152,22 +131,22 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy
return endpoints; return endpoints;
} }
public int getReplicationFactor(String table) public int getReplicationFactor()
{ {
int total = 0; int total = 0;
for (int repFactor : datacenters.get(table).values()) for (int repFactor : datacenters.values())
total += repFactor; total += repFactor;
return total; return total;
} }
public int getReplicationFactor(String dc, String table) public int getReplicationFactor(String dc)
{ {
return datacenters.get(table).get(dc); return datacenters.get(dc);
} }
public Set<String> getDatacenters(String table) public Set<String> getDatacenters()
{ {
return datacenters.get(table).keySet(); return datacenters.keySet();
} }
/** /**
@ -177,7 +156,7 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy
* return a DCQRH with a map of all the DC rep factor. * return a DCQRH with a map of all the DC rep factor.
*/ */
@Override @Override
public AbstractWriteResponseHandler getWriteResponseHandler(Collection<InetAddress> writeEndpoints, Multimap<InetAddress, InetAddress> hintedEndpoints, ConsistencyLevel consistency_level, String table) public AbstractWriteResponseHandler getWriteResponseHandler(Collection<InetAddress> writeEndpoints, Multimap<InetAddress, InetAddress> hintedEndpoints, ConsistencyLevel consistency_level)
{ {
if (consistency_level == ConsistencyLevel.DCQUORUM) if (consistency_level == ConsistencyLevel.DCQUORUM)
{ {
@ -188,7 +167,7 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy
{ {
return new DatacenterSyncWriteResponseHandler(writeEndpoints, hintedEndpoints, consistency_level, table); return new DatacenterSyncWriteResponseHandler(writeEndpoints, hintedEndpoints, consistency_level, table);
} }
return super.getWriteResponseHandler(writeEndpoints, hintedEndpoints, consistency_level, table); return super.getWriteResponseHandler(writeEndpoints, hintedEndpoints, consistency_level);
} }
/** /**
@ -196,12 +175,12 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy
* level is DCQUORUM/DCQUORUMSYNC then it will return a DCQRH. * level is DCQUORUM/DCQUORUMSYNC then it will return a DCQRH.
*/ */
@Override @Override
public QuorumResponseHandler getQuorumResponseHandler(IResponseResolver responseResolver, ConsistencyLevel consistencyLevel, String table) public QuorumResponseHandler getQuorumResponseHandler(IResponseResolver responseResolver, ConsistencyLevel consistencyLevel)
{ {
if (consistencyLevel.equals(ConsistencyLevel.DCQUORUM) || consistencyLevel.equals(ConsistencyLevel.DCQUORUMSYNC)) if (consistencyLevel.equals(ConsistencyLevel.DCQUORUM) || consistencyLevel.equals(ConsistencyLevel.DCQUORUMSYNC))
{ {
return new DatacenterQuorumResponseHandler(responseResolver, consistencyLevel, table); return new DatacenterQuorumResponseHandler(responseResolver, consistencyLevel, table);
} }
return super.getQuorumResponseHandler(responseResolver, consistencyLevel, table); return super.getQuorumResponseHandler(responseResolver, consistencyLevel);
} }
} }

7
src/java/org/apache/cassandra/locator/LocalStrategy.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.locator;
import java.net.InetAddress; import java.net.InetAddress;
import java.util.HashSet; import java.util.HashSet;
import java.util.Map;
import java.util.Set; import java.util.Set;
import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.FBUtilities;
@ -28,12 +29,12 @@ import org.apache.cassandra.dht.Token;
public class LocalStrategy extends AbstractReplicationStrategy public class LocalStrategy extends AbstractReplicationStrategy
{ {
public LocalStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) public LocalStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
{ {
super(tokenMetadata, snitch); super(table, tokenMetadata, snitch, configOptions);
} }
public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata)
{ {
Set<InetAddress> endpoints = new HashSet<InetAddress>(1); Set<InetAddress> endpoints = new HashSet<InetAddress>(1);
InetAddress local = FBUtilities.getLocalAddress(); InetAddress local = FBUtilities.getLocalAddress();

13
src/java/org/apache/cassandra/locator/RackAwareStrategy.java
View File

@ -20,10 +20,7 @@
package org.apache.cassandra.locator; package org.apache.cassandra.locator;
import java.net.InetAddress; import java.net.InetAddress;
import java.util.ArrayList; import java.util.*;
import java.util.LinkedHashSet;
import java.util.Iterator;
import java.util.Set;
import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Token; import org.apache.cassandra.dht.Token;
@ -37,16 +34,16 @@ import org.apache.cassandra.dht.Token;
*/ */
public class RackAwareStrategy extends AbstractReplicationStrategy public class RackAwareStrategy extends AbstractReplicationStrategy
{ {
public RackAwareStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) public RackAwareStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
{ {
super(tokenMetadata, snitch); super(table, tokenMetadata, snitch, configOptions);
if (!(snitch instanceof AbstractRackAwareSnitch)) if (!(snitch instanceof AbstractRackAwareSnitch))
throw new IllegalArgumentException(("RackAwareStrategy requires AbstractRackAwareSnitch.")); throw new IllegalArgumentException(("RackAwareStrategy requires AbstractRackAwareSnitch."));
} }
public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata)
{ {
int replicas = getReplicationFactor(table); int replicas = getReplicationFactor();
Set<InetAddress> endpoints = new LinkedHashSet<InetAddress>(replicas); Set<InetAddress> endpoints = new LinkedHashSet<InetAddress>(replicas);
ArrayList<Token> tokens = metadata.sortedTokens(); ArrayList<Token> tokens = metadata.sortedTokens();

15
src/java/org/apache/cassandra/locator/RackUnawareStrategy.java
View File

@ -20,12 +20,8 @@
package org.apache.cassandra.locator; package org.apache.cassandra.locator;
import java.net.InetAddress; import java.net.InetAddress;
import java.util.ArrayList; import java.util.*;
import java.util.LinkedHashSet;
import java.util.Iterator;
import java.util.Set;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Token; import org.apache.cassandra.dht.Token;
/** /**
@ -36,14 +32,14 @@ import org.apache.cassandra.dht.Token;
*/ */
public class RackUnawareStrategy extends AbstractReplicationStrategy public class RackUnawareStrategy extends AbstractReplicationStrategy
{ {
public RackUnawareStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) public RackUnawareStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
{ {
super(tokenMetadata, snitch); super(table, tokenMetadata, snitch, configOptions);
} }
public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata)
{ {
int replicas = getReplicationFactor(table); int replicas = getReplicationFactor();
ArrayList<Token> tokens = metadata.sortedTokens(); ArrayList<Token> tokens = metadata.sortedTokens();
Set<InetAddress> endpoints = new LinkedHashSet<InetAddress>(replicas); Set<InetAddress> endpoints = new LinkedHashSet<InetAddress>(replicas);
@ -62,4 +58,5 @@ public class RackUnawareStrategy extends AbstractReplicationStrategy
return endpoints; return endpoints;
} }
} }

2
src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java
View File

@ -45,6 +45,6 @@ public class DatacenterQuorumResponseHandler<T> extends QuorumResponseHandler<T>
public int determineBlockFor(ConsistencyLevel consistency_level, String table) public int determineBlockFor(ConsistencyLevel consistency_level, String table)
{ {
DatacenterShardStrategy stategy = (DatacenterShardStrategy) StorageService.instance.getReplicationStrategy(table); DatacenterShardStrategy stategy = (DatacenterShardStrategy) StorageService.instance.getReplicationStrategy(table);
return (stategy.getReplicationFactor(localdc, table) / 2) + 1; return (stategy.getReplicationFactor(localdc) / 2) + 1;
} }
} }

14
src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java
View File

@ -56,7 +56,6 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan
private final DatacenterShardStrategy strategy; private final DatacenterShardStrategy strategy;
private HashMap<String, AtomicInteger> responses = new HashMap<String, AtomicInteger>(); private HashMap<String, AtomicInteger> responses = new HashMap<String, AtomicInteger>();
private final String table;
public DatacenterSyncWriteResponseHandler(Collection<InetAddress> writeEndpoints, Multimap<InetAddress, InetAddress> hintedEndpoints, ConsistencyLevel consistencyLevel, String table) public DatacenterSyncWriteResponseHandler(Collection<InetAddress> writeEndpoints, Multimap<InetAddress, InetAddress> hintedEndpoints, ConsistencyLevel consistencyLevel, String table)
{ {
@ -64,12 +63,11 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan
super(writeEndpoints, hintedEndpoints, consistencyLevel); super(writeEndpoints, hintedEndpoints, consistencyLevel);
assert consistencyLevel == ConsistencyLevel.DCQUORUM; assert consistencyLevel == ConsistencyLevel.DCQUORUM;
this.table = table;
strategy = (DatacenterShardStrategy) StorageService.instance.getReplicationStrategy(table); strategy = (DatacenterShardStrategy) StorageService.instance.getReplicationStrategy(table);
for (String dc : strategy.getDatacenters(table)) for (String dc : strategy.getDatacenters())
{ {
int rf = strategy.getReplicationFactor(dc, table); int rf = strategy.getReplicationFactor(dc);
responses.put(dc, new AtomicInteger((rf / 2) + 1)); responses.put(dc, new AtomicInteger((rf / 2) + 1));
} }
} }
@ -88,14 +86,14 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan
return; return;
} }
// all the quorum conditionas are met // all the quorum conditions are met
condition.signal(); condition.signal();
} }
public void assureSufficientLiveNodes() throws UnavailableException public void assureSufficientLiveNodes() throws UnavailableException
{ {
Map<String, AtomicInteger> dcEndpoints = new HashMap<String, AtomicInteger>(); Map<String, AtomicInteger> dcEndpoints = new HashMap<String, AtomicInteger>();
for (String dc: strategy.getDatacenters(table)) for (String dc: strategy.getDatacenters())
dcEndpoints.put(dc, new AtomicInteger()); dcEndpoints.put(dc, new AtomicInteger());
for (InetAddress destination : hintedEndpoints.keySet()) for (InetAddress destination : hintedEndpoints.keySet())
{ {
@ -105,8 +103,8 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan
dcEndpoints.get(destinationDC).incrementAndGet(); dcEndpoints.get(destinationDC).incrementAndGet();
} }
// Throw exception if any of the DC doesnt have livenodes to accept write. // Throw exception if any of the DC doesn't have livenodes to accept write.
for (String dc: strategy.getDatacenters(table)) for (String dc: strategy.getDatacenters())
{ {
if (dcEndpoints.get(dc).get() != responses.get(dc).get()) if (dcEndpoints.get(dc).get() != responses.get(dc).get())
throw new UnavailableException(); throw new UnavailableException();

4
src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java
View File

@ -63,7 +63,7 @@ public class DatacenterWriteResponseHandler extends WriteResponseHandler
protected int determineBlockFor(String table) protected int determineBlockFor(String table)
{ {
DatacenterShardStrategy strategy = (DatacenterShardStrategy) StorageService.instance.getReplicationStrategy(table); DatacenterShardStrategy strategy = (DatacenterShardStrategy) StorageService.instance.getReplicationStrategy(table);
return (strategy.getReplicationFactor(localdc, table) / 2) + 1; return (strategy.getReplicationFactor(localdc) / 2) + 1;
} }
@ -92,4 +92,4 @@ public class DatacenterWriteResponseHandler extends WriteResponseHandler
throw new UnavailableException(); throw new UnavailableException();
} }
} }
} }

10
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -200,7 +200,7 @@ public class StorageProxy implements StorageProxyMBean
Multimap<InetAddress, InetAddress> hintedEndpoints = rs.getHintedEndpoints(writeEndpoints); Multimap<InetAddress, InetAddress> hintedEndpoints = rs.getHintedEndpoints(writeEndpoints);
// send out the writes, as in mutate() above, but this time with a callback that tracks responses // send out the writes, as in mutate() above, but this time with a callback that tracks responses
final AbstractWriteResponseHandler responseHandler = rs.getWriteResponseHandler(writeEndpoints, hintedEndpoints, consistency_level, table); final AbstractWriteResponseHandler responseHandler = rs.getWriteResponseHandler(writeEndpoints, hintedEndpoints, consistency_level);
responseHandler.assureSufficientLiveNodes(); responseHandler.assureSufficientLiveNodes();
responseHandlers.add(responseHandler); responseHandlers.add(responseHandler);
@ -421,7 +421,7 @@ public class StorageProxy implements StorageProxyMBean
logger.debug("strongread reading " + (m == message ? "data" : "digest") + " for " + command + " from " + m.getMessageId() + "@" + endpoint); logger.debug("strongread reading " + (m == message ? "data" : "digest") + " for " + command + " from " + m.getMessageId() + "@" + endpoint);
} }
AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.table); AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.table);
QuorumResponseHandler<Row> quorumResponseHandler = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), consistency_level, command.table); QuorumResponseHandler<Row> quorumResponseHandler = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), consistency_level);
MessagingService.instance.sendRR(messages, endpoints, quorumResponseHandler); MessagingService.instance.sendRR(messages, endpoints, quorumResponseHandler);
quorumResponseHandlers.add(quorumResponseHandler); quorumResponseHandlers.add(quorumResponseHandler);
commandEndpoints.add(endpoints); commandEndpoints.add(endpoints);
@ -449,7 +449,7 @@ public class StorageProxy implements StorageProxyMBean
if (randomlyReadRepair(command)) if (randomlyReadRepair(command))
{ {
AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.table); AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.table);
QuorumResponseHandler<Row> qrhRepair = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), ConsistencyLevel.QUORUM, command.table); QuorumResponseHandler<Row> qrhRepair = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), ConsistencyLevel.QUORUM);
if (logger.isDebugEnabled()) if (logger.isDebugEnabled())
logger.debug("Digest mismatch:", ex); logger.debug("Digest mismatch:", ex);
Message messageRepair = command.makeReadMessage(); Message messageRepair = command.makeReadMessage();
@ -530,7 +530,7 @@ public class StorageProxy implements StorageProxyMBean
// collect replies and resolve according to consistency level // collect replies and resolve according to consistency level
RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints); RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints);
AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.keyspace); AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.keyspace);
QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver, consistency_level, command.keyspace); QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver, consistency_level);
// TODO bail early if live endpoints can't satisfy requested // TODO bail early if live endpoints can't satisfy requested
// consistency level // consistency level
for (InetAddress endpoint : liveEndpoints) for (InetAddress endpoint : liveEndpoints)
@ -795,7 +795,7 @@ public class StorageProxy implements StorageProxyMBean
Message message = command.getMessage(); Message message = command.getMessage();
RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, endpoints); RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, endpoints);
AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.keyspace); AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.keyspace);
QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver, consistency_level, command.keyspace); QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver, consistency_level);
MessagingService.instance.sendRR(message, endpoints.get(0), handler); MessagingService.instance.sendRR(message, endpoints.get(0), handler);
try try
{ {

44
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -21,7 +21,6 @@ package org.apache.cassandra.service;
import java.io.IOError; import java.io.IOError;
import java.io.IOException; import java.io.IOException;
import java.lang.management.ManagementFactory; import java.lang.management.ManagementFactory;
import java.lang.reflect.Constructor;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.UnknownHostException; import java.net.UnknownHostException;
import java.util.*; import java.util.*;
@ -62,7 +61,6 @@ import org.apache.cassandra.io.DeletionService;
import org.apache.cassandra.io.sstable.SSTableReader; import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.TokenMetadata; import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.ResponseVerbHandler; import org.apache.cassandra.net.ResponseVerbHandler;
@ -274,7 +272,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
public void initReplicationStrategy(String table) public void initReplicationStrategy(String table)
{ {
AbstractReplicationStrategy strat = getReplicationStrategy(tokenMetadata_, table); AbstractReplicationStrategy strat = createReplicationStrategy(tokenMetadata_, table);
replicationStrategies.put(table, strat); replicationStrategies.put(table, strat);
} }
@ -283,17 +281,19 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
replicationStrategies.remove(table); replicationStrategies.remove(table);
} }
public static AbstractReplicationStrategy getReplicationStrategy(TokenMetadata tokenMetadata, String table) public static AbstractReplicationStrategy createReplicationStrategy(TokenMetadata tokenMetadata, String table)
{ {
AbstractReplicationStrategy replicationStrategy = null; AbstractReplicationStrategy replicationStrategy;
Class<? extends AbstractReplicationStrategy> cls = DatabaseDescriptor.getReplicaPlacementStrategyClass(table); KSMetaData ksm = DatabaseDescriptor.getKSMetaData(table);
if (cls == null)
throw new RuntimeException(String.format("No replica strategy configured for %s", table));
Class [] parameterTypes = new Class[] { TokenMetadata.class, IEndpointSnitch.class};
try try
{ {
Constructor<? extends AbstractReplicationStrategy> constructor = cls.getConstructor(parameterTypes); replicationStrategy = AbstractReplicationStrategy.createReplicationStrategy(
replicationStrategy = constructor.newInstance(tokenMetadata, DatabaseDescriptor.getEndpointSnitch()); table,
ksm.strategyClass,
tokenMetadata,
DatabaseDescriptor.getEndpointSnitch(),
ksm.strategyOptions
);
} }
catch (Exception e) catch (Exception e)
{ {
@ -543,7 +543,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
Map<Range, List<InetAddress>> rangeToEndpointMap = new HashMap<Range, List<InetAddress>>(); Map<Range, List<InetAddress>> rangeToEndpointMap = new HashMap<Range, List<InetAddress>>();
for (Range range : ranges) for (Range range : ranges)
{ {
rangeToEndpointMap.put(range, getReplicationStrategy(keyspace).getNaturalEndpoints(range.right, keyspace)); rangeToEndpointMap.put(range, getReplicationStrategy(keyspace).getNaturalEndpoints(range.right));
} }
return rangeToEndpointMap; return rangeToEndpointMap;
} }
@ -785,7 +785,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
return; return;
} }
Multimap<InetAddress, Range> addressRanges = strategy.getAddressRanges(table); Multimap<InetAddress, Range> addressRanges = strategy.getAddressRanges();
// Copy of metadata reflecting the situation after all leave operations are finished. // Copy of metadata reflecting the situation after all leave operations are finished.
TokenMetadata allLeftMetadata = tm.cloneAfterAllLeft(); TokenMetadata allLeftMetadata = tm.cloneAfterAllLeft();
@ -799,8 +799,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
// all leaving nodes are gone. // all leaving nodes are gone.
for (Range range : affectedRanges) for (Range range : affectedRanges)
{ {
Set<InetAddress> currentEndpoints = strategy.calculateNaturalEndpoints(range.right, tm, table); Set<InetAddress> currentEndpoints = strategy.calculateNaturalEndpoints(range.right, tm);
Set<InetAddress> newEndpoints = strategy.calculateNaturalEndpoints(range.right, allLeftMetadata, table); Set<InetAddress> newEndpoints = strategy.calculateNaturalEndpoints(range.right, allLeftMetadata);
newEndpoints.removeAll(currentEndpoints); newEndpoints.removeAll(currentEndpoints);
pendingRanges.putAll(range, newEndpoints); pendingRanges.putAll(range, newEndpoints);
} }
@ -815,7 +815,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
InetAddress endpoint = entry.getValue(); InetAddress endpoint = entry.getValue();
allLeftMetadata.updateNormalToken(entry.getKey(), endpoint); allLeftMetadata.updateNormalToken(entry.getKey(), endpoint);
for (Range range : strategy.getAddressRanges(allLeftMetadata, table).get(endpoint)) for (Range range : strategy.getAddressRanges(allLeftMetadata).get(endpoint))
pendingRanges.put(range, endpoint); pendingRanges.put(range, endpoint);
allLeftMetadata.removeEndpoint(endpoint); allLeftMetadata.removeEndpoint(endpoint);
} }
@ -860,7 +860,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
if (logger_.isDebugEnabled()) if (logger_.isDebugEnabled())
logger_.debug(endpoint + " was removed, my added ranges: " + StringUtils.join(myNewRanges, ", ")); logger_.debug(endpoint + " was removed, my added ranges: " + StringUtils.join(myNewRanges, ", "));
Multimap<Range, InetAddress> rangeAddresses = getReplicationStrategy(table).getRangeAddresses(tokenMetadata_, table); Multimap<Range, InetAddress> rangeAddresses = getReplicationStrategy(table).getRangeAddresses(tokenMetadata_);
Multimap<InetAddress, Range> sourceRanges = HashMultimap.create(); Multimap<InetAddress, Range> sourceRanges = HashMultimap.create();
IFailureDetector failureDetector = FailureDetector.instance; IFailureDetector failureDetector = FailureDetector.instance;
@ -909,7 +909,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
// Find (for each range) all nodes that store replicas for these ranges as well // Find (for each range) all nodes that store replicas for these ranges as well
for (Range range : ranges) for (Range range : ranges)
currentReplicaEndpoints.put(range, getReplicationStrategy(table).calculateNaturalEndpoints(range.right, tokenMetadata_, table)); currentReplicaEndpoints.put(range, getReplicationStrategy(table).calculateNaturalEndpoints(range.right, tokenMetadata_));
TokenMetadata temp = tokenMetadata_.cloneAfterAllLeft(); TokenMetadata temp = tokenMetadata_.cloneAfterAllLeft();
@ -927,7 +927,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
// range. // range.
for (Range range : ranges) for (Range range : ranges)
{ {
Set<InetAddress> newReplicaEndpoints = getReplicationStrategy(table).calculateNaturalEndpoints(range.right, temp, table); Set<InetAddress> newReplicaEndpoints = getReplicationStrategy(table).calculateNaturalEndpoints(range.right, temp);
newReplicaEndpoints.removeAll(currentReplicaEndpoints.get(range)); newReplicaEndpoints.removeAll(currentReplicaEndpoints.get(range));
if (logger_.isDebugEnabled()) if (logger_.isDebugEnabled())
if (newReplicaEndpoints.isEmpty()) if (newReplicaEndpoints.isEmpty())
@ -1227,7 +1227,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
*/ */
Collection<Range> getRangesForEndpoint(String table, InetAddress ep) Collection<Range> getRangesForEndpoint(String table, InetAddress ep)
{ {
return getReplicationStrategy(table).getAddressRanges(table).get(ep); return getReplicationStrategy(table).getAddressRanges().get(ep);
} }
/** /**
@ -1277,7 +1277,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
*/ */
public List<InetAddress> getNaturalEndpoints(String table, Token token) public List<InetAddress> getNaturalEndpoints(String table, Token token)
{ {
return getReplicationStrategy(table).getNaturalEndpoints(token, table); return getReplicationStrategy(table).getNaturalEndpoints(token);
} }
/** /**
@ -1295,7 +1295,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
public List<InetAddress> getLiveNaturalEndpoints(String table, Token token) public List<InetAddress> getLiveNaturalEndpoints(String table, Token token)
{ {
List<InetAddress> liveEps = new ArrayList<InetAddress>(); List<InetAddress> liveEps = new ArrayList<InetAddress>();
List<InetAddress> endpoints = getReplicationStrategy(table).getNaturalEndpoints(token, table); List<InetAddress> endpoints = getReplicationStrategy(table).getNaturalEndpoints(token);
for (InetAddress endpoint : endpoints) for (InetAddress endpoint : endpoints)
{ {

5
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -889,8 +889,9 @@ public class CassandraServer implements Cassandra.Iface
KSMetaData ksm = new KSMetaData( KSMetaData ksm = new KSMetaData(
ks_def.name, ks_def.name,
(Class<? extends AbstractReplicationStrategy>)Class.forName(ks_def.strategy_class), (Class<? extends AbstractReplicationStrategy>)Class.forName(ks_def.strategy_class),
ks_def.replication_factor, ks_def.strategy_options,
ks_def.replication_factor,
cfDefs.toArray(new CFMetaData[cfDefs.size()])); cfDefs.toArray(new CFMetaData[cfDefs.size()]));
applyMigrationOnStage(new AddKeyspace(ksm)); applyMigrationOnStage(new AddKeyspace(ksm));
return DatabaseDescriptor.getDefsVersion().toString(); return DatabaseDescriptor.getDefsVersion().toString();

8
test/system/__init__.py
View File

@ -142,7 +142,7 @@ class ThriftTester(BaseTester):
self.client.transport.close() self.client.transport.close()
def define_schema(self): def define_schema(self):
keyspace1 = Cassandra.KsDef('Keyspace1', 'org.apache.cassandra.locator.RackUnawareStrategy', 1, keyspace1 = Cassandra.KsDef('Keyspace1', 'org.apache.cassandra.locator.RackUnawareStrategy', None, 1,
[ [
Cassandra.CfDef('Keyspace1', 'Standard1'), Cassandra.CfDef('Keyspace1', 'Standard1'),
Cassandra.CfDef('Keyspace1', 'Standard2'), Cassandra.CfDef('Keyspace1', 'Standard2'),
@ -156,7 +156,7 @@ class ThriftTester(BaseTester):
Cassandra.CfDef('Keyspace1', 'Indexed1', column_metadata=[Cassandra.ColumnDef('birthdate', 'LongType', Cassandra.IndexType.KEYS, 'birthdate')]), Cassandra.CfDef('Keyspace1', 'Indexed1', column_metadata=[Cassandra.ColumnDef('birthdate', 'LongType', Cassandra.IndexType.KEYS, 'birthdate')]),
]) ])
keyspace2 = Cassandra.KsDef('Keyspace2', 'org.apache.cassandra.locator.RackUnawareStrategy', 1, keyspace2 = Cassandra.KsDef('Keyspace2', 'org.apache.cassandra.locator.RackUnawareStrategy', None, 1,
[ [
Cassandra.CfDef('Keyspace2', 'Standard1'), Cassandra.CfDef('Keyspace2', 'Standard1'),
Cassandra.CfDef('Keyspace2', 'Standard3'), Cassandra.CfDef('Keyspace2', 'Standard3'),
@ -164,12 +164,12 @@ class ThriftTester(BaseTester):
Cassandra.CfDef('Keyspace2', 'Super4', column_type='Super', subcomparator_type='TimeUUIDType'), Cassandra.CfDef('Keyspace2', 'Super4', column_type='Super', subcomparator_type='TimeUUIDType'),
]) ])
keyspace3 = Cassandra.KsDef('Keyspace3', 'org.apache.cassandra.locator.RackUnawareStrategy', 5, keyspace3 = Cassandra.KsDef('Keyspace3', 'org.apache.cassandra.locator.RackUnawareStrategy', None, 5,
[ [
Cassandra.CfDef('Keyspace3', 'Standard1'), Cassandra.CfDef('Keyspace3', 'Standard1'),
]) ])
keyspace4 = Cassandra.KsDef('Keyspace4', 'org.apache.cassandra.locator.RackUnawareStrategy', 3, keyspace4 = Cassandra.KsDef('Keyspace4', 'org.apache.cassandra.locator.RackUnawareStrategy', None, 3,
[ [
Cassandra.CfDef('Keyspace4', 'Standard1'), Cassandra.CfDef('Keyspace4', 'Standard1'),
Cassandra.CfDef('Keyspace4', 'Standard3'), Cassandra.CfDef('Keyspace4', 'Standard3'),

2
test/system/test_thrift_server.py
View File

@ -1124,7 +1124,7 @@ class TestMutations(ThriftTester):
def test_system_keyspace_operations(self): def test_system_keyspace_operations(self):
""" Test keyspace (add, drop, rename) operations """ """ Test keyspace (add, drop, rename) operations """
# create # create
keyspace = KsDef('CreateKeyspace', 'org.apache.cassandra.locator.RackUnawareStrategy', 1, keyspace = KsDef('CreateKeyspace', 'org.apache.cassandra.locator.RackUnawareStrategy', {}, 1,
[CfDef('CreateKeyspace', 'CreateKsCf')]) [CfDef('CreateKeyspace', 'CreateKsCf')])
client.system_add_keyspace(keyspace) client.system_add_keyspace(keyspace)
newks = client.describe_keyspace('CreateKeyspace') newks = client.describe_keyspace('CreateKeyspace')

4
test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java
View File

@ -83,9 +83,9 @@ public class DatabaseDescriptorTest
assert DatabaseDescriptor.getNonSystemTables().size() == 0; assert DatabaseDescriptor.getNonSystemTables().size() == 0;
// add a few. // add a few.
AddKeyspace ks0 = new AddKeyspace(new KSMetaData("ks0", RackUnawareStrategy.class, 3)); AddKeyspace ks0 = new AddKeyspace(new KSMetaData("ks0", RackUnawareStrategy.class, null, 3));
ks0.apply(); ks0.apply();
AddKeyspace ks1 = new AddKeyspace(new KSMetaData("ks1", RackUnawareStrategy.class, 3)); AddKeyspace ks1 = new AddKeyspace(new KSMetaData("ks1", RackUnawareStrategy.class, null, 3));
ks1.apply(); ks1.apply();
assert DatabaseDescriptor.getTableDefinition("ks0") != null; assert DatabaseDescriptor.getTableDefinition("ks0") != null;

6
test/unit/org/apache/cassandra/db/DefsTest.java
View File

@ -66,7 +66,7 @@ public class DefsTest extends CleanupHelper
try try
{ {
new AddColumnFamily(newCf).apply(); new AddColumnFamily(newCf).apply();
throw new AssertionError("You should't be able to do anything to a keyspace that doesn't exist."); throw new AssertionError("You shouldn't be able to do anything to a keyspace that doesn't exist.");
} }
catch (ConfigurationException expected) catch (ConfigurationException expected)
{ {
@ -262,7 +262,7 @@ public class DefsTest extends CleanupHelper
{ {
DecoratedKey dk = Util.dk("key0"); DecoratedKey dk = Util.dk("key0");
CFMetaData newCf = new CFMetaData("NewKeyspace1", "AddedStandard1", ColumnFamilyType.Standard, ClockType.Timestamp, UTF8Type.instance, null, new TimestampReconciler(), "A new cf for a new ks", 0, false, 1.0, 0, 864000, Collections.<byte[], ColumnDefinition>emptyMap()); CFMetaData newCf = new CFMetaData("NewKeyspace1", "AddedStandard1", ColumnFamilyType.Standard, ClockType.Timestamp, UTF8Type.instance, null, new TimestampReconciler(), "A new cf for a new ks", 0, false, 1.0, 0, 864000, Collections.<byte[], ColumnDefinition>emptyMap());
KSMetaData newKs = new KSMetaData(newCf.tableName, RackUnawareStrategy.class, 5, newCf); KSMetaData newKs = new KSMetaData(newCf.tableName, RackUnawareStrategy.class, null, 5, newCf);
new AddKeyspace(newKs).apply(); new AddKeyspace(newKs).apply();
@ -414,7 +414,7 @@ public class DefsTest extends CleanupHelper
{ {
assert DatabaseDescriptor.getTableDefinition("EmptyKeyspace") == null; assert DatabaseDescriptor.getTableDefinition("EmptyKeyspace") == null;
KSMetaData newKs = new KSMetaData("EmptyKeyspace", RackUnawareStrategy.class, 5, new CFMetaData[]{}); KSMetaData newKs = new KSMetaData("EmptyKeyspace", RackUnawareStrategy.class, null, 5);
new AddKeyspace(newKs).apply(); new AddKeyspace(newKs).apply();
assert DatabaseDescriptor.getTableDefinition("EmptyKeyspace") != null; assert DatabaseDescriptor.getTableDefinition("EmptyKeyspace") != null;

22
test/unit/org/apache/cassandra/locator/DatacenterShardStrategyTest.java
View File

@ -23,7 +23,9 @@ import java.io.IOException;
import java.net.InetAddress; import java.net.InetAddress;
import java.net.UnknownHostException; import java.net.UnknownHostException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet; import java.util.HashSet;
import java.util.Map;
import javax.xml.parsers.ParserConfigurationException; import javax.xml.parsers.ParserConfigurationException;
import org.junit.Test; import org.junit.Test;
@ -40,16 +42,22 @@ public class DatacenterShardStrategyTest
@Test @Test
public void testProperties() throws IOException, ParserConfigurationException, SAXException, ConfigurationException public void testProperties() throws IOException, ParserConfigurationException, SAXException, ConfigurationException
{ {
PropertyFileSnitch snitch = new PropertyFileSnitch(); IEndpointSnitch snitch = new PropertyFileSnitch();
TokenMetadata metadata = new TokenMetadata(); TokenMetadata metadata = new TokenMetadata();
createDummyTokens(metadata); createDummyTokens(metadata);
// Set the localhost to the tokenmetadata. Embeded cassandra way?
DatacenterShardStrategy strategy = new DatacenterShardStrategy(metadata, snitch); Map<String, String> configOptions = new HashMap<String, String>();
assert strategy.getReplicationFactor("DC1", table) == 3; configOptions.put("DC1", "3");
assert strategy.getReplicationFactor("DC2", table) == 2; configOptions.put("DC2", "2");
assert strategy.getReplicationFactor("DC3", table) == 1; configOptions.put("DC3", "1");
// Set the localhost to the tokenmetadata. Embedded cassandra way?
DatacenterShardStrategy strategy = new DatacenterShardStrategy(table, metadata, snitch, configOptions);
assert strategy.getReplicationFactor("DC1") == 3;
assert strategy.getReplicationFactor("DC2") == 2;
assert strategy.getReplicationFactor("DC3") == 1;
// Query for the natural hosts // Query for the natural hosts
ArrayList<InetAddress> endpoints = strategy.getNaturalEndpoints(new StringToken("123"), table); ArrayList<InetAddress> endpoints = strategy.getNaturalEndpoints(new StringToken("123"));
assert 6 == endpoints.size(); assert 6 == endpoints.size();
assert 6 == new HashSet<InetAddress>(endpoints).size(); // ensure uniqueness assert 6 == new HashSet<InetAddress>(endpoints).size(); // ensure uniqueness
} }

8
test/unit/org/apache/cassandra/locator/RackAwareStrategyTest.java
View File

@ -60,7 +60,7 @@ public class RackAwareStrategyTest
{ {
RackInferringSnitch endpointSnitch = new RackInferringSnitch(); RackInferringSnitch endpointSnitch = new RackInferringSnitch();
AbstractReplicationStrategy strategy = new RackAwareStrategy(tmd, endpointSnitch); AbstractReplicationStrategy strategy = new RackAwareStrategy("Keyspace1", tmd, endpointSnitch, null);
addEndpoint("0", "5", "254.0.0.1"); addEndpoint("0", "5", "254.0.0.1");
addEndpoint("10", "15", "254.0.0.2"); addEndpoint("10", "15", "254.0.0.2");
addEndpoint("20", "25", "254.0.0.3"); addEndpoint("20", "25", "254.0.0.3");
@ -85,7 +85,7 @@ public class RackAwareStrategyTest
{ {
RackInferringSnitch endpointSnitch = new RackInferringSnitch(); RackInferringSnitch endpointSnitch = new RackInferringSnitch();
AbstractReplicationStrategy strategy = new RackAwareStrategy(tmd, endpointSnitch); AbstractReplicationStrategy strategy = new RackAwareStrategy("Keyspace1", tmd, endpointSnitch, null);
addEndpoint("0", "5", "254.0.0.1"); addEndpoint("0", "5", "254.0.0.1");
addEndpoint("10", "15", "254.0.0.2"); addEndpoint("10", "15", "254.0.0.2");
addEndpoint("20", "25", "254.1.0.3"); addEndpoint("20", "25", "254.1.0.3");
@ -111,7 +111,7 @@ public class RackAwareStrategyTest
{ {
RackInferringSnitch endpointSnitch = new RackInferringSnitch(); RackInferringSnitch endpointSnitch = new RackInferringSnitch();
AbstractReplicationStrategy strategy = new RackAwareStrategy(tmd, endpointSnitch); AbstractReplicationStrategy strategy = new RackAwareStrategy("Keyspace1", tmd, endpointSnitch, null);
addEndpoint("0", "5", "254.0.0.1"); addEndpoint("0", "5", "254.0.0.1");
addEndpoint("10", "15", "254.0.0.2"); addEndpoint("10", "15", "254.0.0.2");
addEndpoint("20", "25", "254.0.1.3"); addEndpoint("20", "25", "254.0.1.3");
@ -160,7 +160,7 @@ public class RackAwareStrategyTest
{ {
for (Token keyToken : keyTokens) for (Token keyToken : keyTokens)
{ {
List<InetAddress> endpoints = strategy.getNaturalEndpoints(keyToken, table); List<InetAddress> endpoints = strategy.getNaturalEndpoints(keyToken);
for (int j = 0; j < endpoints.size(); j++) for (int j = 0; j < endpoints.size(); j++)
{ {
ArrayList<InetAddress> hostsExpected = expectedResults.get(keyToken.toString()); ArrayList<InetAddress> hostsExpected = expectedResults.get(keyToken.toString());

53
test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java
View File

@ -26,6 +26,7 @@ import java.util.Collection;
import java.util.HashSet; import java.util.HashSet;
import java.util.List; import java.util.List;
import org.apache.cassandra.config.ConfigurationException;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*; import static org.junit.Assert.*;
@ -45,7 +46,7 @@ public class RackUnawareStrategyTest extends SchemaLoader
try try
{ {
rs = StorageService.instance.getReplicationStrategy("SomeBogusTableThatDoesntExist"); rs = StorageService.instance.getReplicationStrategy("SomeBogusTableThatDoesntExist");
throw new AssertionError("SS.getReplicationStrategy() should have thrown a RuntimeException."); throw new AssertionError("SS.createReplicationStrategy() should have thrown a RuntimeException.");
} }
catch (RuntimeException ex) catch (RuntimeException ex)
{ {
@ -54,26 +55,21 @@ public class RackUnawareStrategyTest extends SchemaLoader
} }
@Test @Test
public void testBigIntegerEndpoints() throws UnknownHostException public void testBigIntegerEndpoints() throws UnknownHostException, ConfigurationException
{ {
TokenMetadata tmd = new TokenMetadata();
AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, new SimpleSnitch());
List<Token> endpointTokens = new ArrayList<Token>(); List<Token> endpointTokens = new ArrayList<Token>();
List<Token> keyTokens = new ArrayList<Token>(); List<Token> keyTokens = new ArrayList<Token>();
for (int i = 0; i < 5; i++) { for (int i = 0; i < 5; i++) {
endpointTokens.add(new BigIntegerToken(String.valueOf(10 * i))); endpointTokens.add(new BigIntegerToken(String.valueOf(10 * i)));
keyTokens.add(new BigIntegerToken(String.valueOf(10 * i + 5))); keyTokens.add(new BigIntegerToken(String.valueOf(10 * i + 5)));
} }
verifyGetNaturalEndpoints(tmd, strategy, endpointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0])); verifyGetNaturalEndpoints(endpointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0]));
} }
@Test @Test
public void testStringEndpoints() throws UnknownHostException public void testStringEndpoints() throws UnknownHostException, ConfigurationException
{ {
TokenMetadata tmd = new TokenMetadata();
IPartitioner partitioner = new OrderPreservingPartitioner(); IPartitioner partitioner = new OrderPreservingPartitioner();
AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, new SimpleSnitch());
List<Token> endpointTokens = new ArrayList<Token>(); List<Token> endpointTokens = new ArrayList<Token>();
List<Token> keyTokens = new ArrayList<Token>(); List<Token> keyTokens = new ArrayList<Token>();
@ -81,15 +77,19 @@ public class RackUnawareStrategyTest extends SchemaLoader
endpointTokens.add(new StringToken(String.valueOf((char)('a' + i * 2)))); endpointTokens.add(new StringToken(String.valueOf((char)('a' + i * 2))));
keyTokens.add(partitioner.getToken(String.valueOf((char)('a' + i * 2 + 1)).getBytes())); keyTokens.add(partitioner.getToken(String.valueOf((char)('a' + i * 2 + 1)).getBytes()));
} }
verifyGetNaturalEndpoints(tmd, strategy, endpointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0])); verifyGetNaturalEndpoints(endpointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0]));
} }
// given a list of endpoint tokens, and a set of key tokens falling between the endpoint tokens, // given a list of endpoint tokens, and a set of key tokens falling between the endpoint tokens,
// make sure that the Strategy picks the right endpoints for the keys. // make sure that the Strategy picks the right endpoints for the keys.
private void verifyGetNaturalEndpoints(TokenMetadata tmd, AbstractReplicationStrategy strategy, Token[] endpointTokens, Token[] keyTokens) throws UnknownHostException private void verifyGetNaturalEndpoints(Token[] endpointTokens, Token[] keyTokens) throws UnknownHostException, ConfigurationException
{ {
TokenMetadata tmd;
AbstractReplicationStrategy strategy;
for (String table : DatabaseDescriptor.getNonSystemTables()) for (String table : DatabaseDescriptor.getNonSystemTables())
{ {
tmd = new TokenMetadata();
strategy = getStrategy(table, tmd);
List<InetAddress> hosts = new ArrayList<InetAddress>(); List<InetAddress> hosts = new ArrayList<InetAddress>();
for (int i = 0; i < endpointTokens.length; i++) for (int i = 0; i < endpointTokens.length; i++)
{ {
@ -100,7 +100,7 @@ public class RackUnawareStrategyTest extends SchemaLoader
for (int i = 0; i < keyTokens.length; i++) for (int i = 0; i < keyTokens.length; i++)
{ {
List<InetAddress> endpoints = strategy.getNaturalEndpoints(keyTokens[i], table); List<InetAddress> endpoints = strategy.getNaturalEndpoints(keyTokens[i]);
assertEquals(DatabaseDescriptor.getReplicationFactor(table), endpoints.size()); assertEquals(DatabaseDescriptor.getReplicationFactor(table), endpoints.size());
List<InetAddress> correctEndpoints = new ArrayList<InetAddress>(); List<InetAddress> correctEndpoints = new ArrayList<InetAddress>();
for (int j = 0; j < endpoints.size(); j++) for (int j = 0; j < endpoints.size(); j++)
@ -111,13 +111,12 @@ public class RackUnawareStrategyTest extends SchemaLoader
} }
@Test @Test
public void testGetEndpointsDuringBootstrap() throws UnknownHostException public void testGetEndpointsDuringBootstrap() throws UnknownHostException, ConfigurationException
{ {
// the token difference will be RING_SIZE * 2. // the token difference will be RING_SIZE * 2.
final int RING_SIZE = 10; final int RING_SIZE = 10;
TokenMetadata tmd = new TokenMetadata(); TokenMetadata tmd = new TokenMetadata();
TokenMetadata oldTmd = StorageServiceAccessor.setTokenMetadata(tmd); TokenMetadata oldTmd = StorageServiceAccessor.setTokenMetadata(tmd);
AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, new SimpleSnitch());
Token[] endpointTokens = new Token[RING_SIZE]; Token[] endpointTokens = new Token[RING_SIZE];
Token[] keyTokens = new Token[RING_SIZE]; Token[] keyTokens = new Token[RING_SIZE];
@ -141,14 +140,18 @@ public class RackUnawareStrategyTest extends SchemaLoader
InetAddress bootstrapEndpoint = InetAddress.getByName("127.0.0.11"); InetAddress bootstrapEndpoint = InetAddress.getByName("127.0.0.11");
tmd.addBootstrapToken(bsToken, bootstrapEndpoint); tmd.addBootstrapToken(bsToken, bootstrapEndpoint);
AbstractReplicationStrategy strategy = null;
for (String table : DatabaseDescriptor.getNonSystemTables()) for (String table : DatabaseDescriptor.getNonSystemTables())
{ {
strategy = getStrategy(table, tmd);
StorageService.calculatePendingRanges(strategy, table); StorageService.calculatePendingRanges(strategy, table);
int replicationFactor = DatabaseDescriptor.getReplicationFactor(table); int replicationFactor = DatabaseDescriptor.getReplicationFactor(table);
for (int i = 0; i < keyTokens.length; i++) for (int i = 0; i < keyTokens.length; i++)
{ {
Collection<InetAddress> endpoints = tmd.getWriteEndpoints(keyTokens[i], table, strategy.getNaturalEndpoints(keyTokens[i], table)); Collection<InetAddress> endpoints = tmd.getWriteEndpoints(keyTokens[i], table, strategy.getNaturalEndpoints(keyTokens[i]));
assertTrue(endpoints.size() >= replicationFactor); assertTrue(endpoints.size() >= replicationFactor);
for (int j = 0; j < replicationFactor; j++) for (int j = 0; j < replicationFactor; j++)
@ -167,4 +170,24 @@ public class RackUnawareStrategyTest extends SchemaLoader
StorageServiceAccessor.setTokenMetadata(oldTmd); StorageServiceAccessor.setTokenMetadata(oldTmd);
} }
private AbstractReplicationStrategy getStrategyWithNewTokenMetadata(AbstractReplicationStrategy strategy, TokenMetadata newTmd) throws ConfigurationException
{
return AbstractReplicationStrategy.createReplicationStrategy(
strategy.table,
strategy.getClass().getName(),
newTmd,
strategy.snitch,
strategy.configOptions);
}
private AbstractReplicationStrategy getStrategy(String table, TokenMetadata tmd) throws ConfigurationException
{
return AbstractReplicationStrategy.createReplicationStrategy(
table,
"org.apache.cassandra.locator.RackUnawareStrategy",
tmd,
new SimpleSnitch(),
null);
}
} }

84
test/unit/org/apache/cassandra/locator/ReplicationStrategyEndpointCacheTest.java
View File

@ -21,10 +21,9 @@ package org.apache.cassandra.locator;
import java.lang.reflect.Constructor; import java.lang.reflect.Constructor;
import java.net.InetAddress; import java.net.InetAddress;
import java.util.ArrayList; import java.util.*;
import java.util.HashSet;
import java.util.Set;
import org.apache.cassandra.service.StorageService;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.junit.Test; import org.junit.Test;
@ -39,12 +38,12 @@ public class ReplicationStrategyEndpointCacheTest extends SchemaLoader
private Token searchToken; private Token searchToken;
private AbstractReplicationStrategy strategy; private AbstractReplicationStrategy strategy;
public void setup(Class stratClass) throws Exception public void setup(Class stratClass, Map<String, String> strategyOptions) throws Exception
{ {
tmd = new TokenMetadata(); tmd = new TokenMetadata();
searchToken = new BigIntegerToken(String.valueOf(15)); searchToken = new BigIntegerToken(String.valueOf(15));
Constructor constructor = stratClass.getConstructor(TokenMetadata.class, IEndpointSnitch.class);
strategy = (AbstractReplicationStrategy) constructor.newInstance(tmd, new PropertyFileSnitch()); strategy = getStrategyWithNewTokenMetadata(StorageService.instance.getReplicationStrategy("Keyspace3"), tmd);
tmd.updateNormalToken(new BigIntegerToken(String.valueOf(10)), InetAddress.getByName("127.0.0.1")); tmd.updateNormalToken(new BigIntegerToken(String.valueOf(10)), InetAddress.getByName("127.0.0.1"));
tmd.updateNormalToken(new BigIntegerToken(String.valueOf(20)), InetAddress.getByName("127.0.0.2")); tmd.updateNormalToken(new BigIntegerToken(String.valueOf(20)), InetAddress.getByName("127.0.0.2"));
@ -59,73 +58,73 @@ public class ReplicationStrategyEndpointCacheTest extends SchemaLoader
@Test @Test
public void testEndpointsWereCached() throws Exception public void testEndpointsWereCached() throws Exception
{ {
runEndpointsWereCachedTest(FakeRackUnawareStrategy.class); runEndpointsWereCachedTest(FakeRackUnawareStrategy.class, null);
runEndpointsWereCachedTest(FakeRackAwareStrategy.class); runEndpointsWereCachedTest(FakeRackAwareStrategy.class, null);
runEndpointsWereCachedTest(FakeDatacenterShardStrategy.class); runEndpointsWereCachedTest(FakeDatacenterShardStrategy.class, new HashMap<String, String>());
} }
public void runEndpointsWereCachedTest(Class stratClass) throws Exception public void runEndpointsWereCachedTest(Class stratClass, Map<String, String> configOptions) throws Exception
{ {
setup(stratClass); setup(stratClass, configOptions);
assert strategy.getNaturalEndpoints(searchToken, "Keyspace3").equals(strategy.getNaturalEndpoints(searchToken, "Keyspace3")); assert strategy.getNaturalEndpoints(searchToken).equals(strategy.getNaturalEndpoints(searchToken));
} }
@Test @Test
public void testCacheRespectsTokenChanges() throws Exception public void testCacheRespectsTokenChanges() throws Exception
{ {
runCacheRespectsTokenChangesTest(RackUnawareStrategy.class); runCacheRespectsTokenChangesTest(RackUnawareStrategy.class, null);
runCacheRespectsTokenChangesTest(RackAwareStrategy.class); runCacheRespectsTokenChangesTest(RackAwareStrategy.class, null);
runCacheRespectsTokenChangesTest(DatacenterShardStrategy.class); runCacheRespectsTokenChangesTest(DatacenterShardStrategy.class, new HashMap<String, String>());
} }
public void runCacheRespectsTokenChangesTest(Class stratClass) throws Exception public void runCacheRespectsTokenChangesTest(Class stratClass, Map<String, String> configOptions) throws Exception
{ {
setup(stratClass); setup(stratClass, configOptions);
ArrayList<InetAddress> initial; ArrayList<InetAddress> initial;
ArrayList<InetAddress> endpoints; ArrayList<InetAddress> endpoints;
endpoints = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); endpoints = strategy.getNaturalEndpoints(searchToken);
assert endpoints.size() == 5 : StringUtils.join(endpoints, ","); assert endpoints.size() == 5 : StringUtils.join(endpoints, ",");
// test token addition, in DC2 before existing token // test token addition, in DC2 before existing token
initial = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); initial = strategy.getNaturalEndpoints(searchToken);
tmd.updateNormalToken(new BigIntegerToken(String.valueOf(35)), InetAddress.getByName("127.0.0.5")); tmd.updateNormalToken(new BigIntegerToken(String.valueOf(35)), InetAddress.getByName("127.0.0.5"));
endpoints = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); endpoints = strategy.getNaturalEndpoints(searchToken);
assert endpoints.size() == 5 : StringUtils.join(endpoints, ","); assert endpoints.size() == 5 : StringUtils.join(endpoints, ",");
assert !endpoints.equals(initial); assert !endpoints.equals(initial);
// test token removal, newly created token // test token removal, newly created token
initial = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); initial = strategy.getNaturalEndpoints(searchToken);
tmd.removeEndpoint(InetAddress.getByName("127.0.0.5")); tmd.removeEndpoint(InetAddress.getByName("127.0.0.5"));
endpoints = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); endpoints = strategy.getNaturalEndpoints(searchToken);
assert endpoints.size() == 5 : StringUtils.join(endpoints, ","); assert endpoints.size() == 5 : StringUtils.join(endpoints, ",");
assert !endpoints.contains(InetAddress.getByName("127.0.0.5")); assert !endpoints.contains(InetAddress.getByName("127.0.0.5"));
assert !endpoints.equals(initial); assert !endpoints.equals(initial);
// test token change // test token change
initial = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); initial = strategy.getNaturalEndpoints(searchToken);
//move .8 after search token but before other DC3 //move .8 after search token but before other DC3
tmd.updateNormalToken(new BigIntegerToken(String.valueOf(25)), InetAddress.getByName("127.0.0.8")); tmd.updateNormalToken(new BigIntegerToken(String.valueOf(25)), InetAddress.getByName("127.0.0.8"));
endpoints = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); endpoints = strategy.getNaturalEndpoints(searchToken);
assert endpoints.size() == 5 : StringUtils.join(endpoints, ","); assert endpoints.size() == 5 : StringUtils.join(endpoints, ",");
assert !endpoints.equals(initial); assert !endpoints.equals(initial);
} }
protected static class FakeRackUnawareStrategy extends RackUnawareStrategy protected static class FakeRackUnawareStrategy extends RackUnawareStrategy
{ {
private boolean called = false; private boolean called = false;
public FakeRackUnawareStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) public FakeRackUnawareStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
{ {
super(tokenMetadata, snitch); super(table, tokenMetadata, snitch, configOptions);
} }
@Override @Override
public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata)
{ {
assert !called : "calculateNaturalEndpoints was already called, result should have been cached"; assert !called : "calculateNaturalEndpoints was already called, result should have been cached";
called = true; called = true;
return super.calculateNaturalEndpoints(token, metadata, table); return super.calculateNaturalEndpoints(token, metadata);
} }
} }
@ -133,17 +132,17 @@ public class ReplicationStrategyEndpointCacheTest extends SchemaLoader
{ {
private boolean called = false; private boolean called = false;
public FakeRackAwareStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) public FakeRackAwareStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions)
{ {
super(tokenMetadata, snitch); super(table, tokenMetadata, snitch, configOptions);
} }
@Override @Override
public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata)
{ {
assert !called : "calculateNaturalEndpoints was already called, result should have been cached"; assert !called : "calculateNaturalEndpoints was already called, result should have been cached";
called = true; called = true;
return super.calculateNaturalEndpoints(token, metadata, table); return super.calculateNaturalEndpoints(token, metadata);
} }
} }
@ -151,17 +150,28 @@ public class ReplicationStrategyEndpointCacheTest extends SchemaLoader
{ {
private boolean called = false; private boolean called = false;
public FakeDatacenterShardStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) throws ConfigurationException public FakeDatacenterShardStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map<String, String> configOptions) throws ConfigurationException
{ {
super(tokenMetadata, snitch); super(table, tokenMetadata, snitch, configOptions);
} }
@Override @Override
public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) public Set<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata)
{ {
assert !called : "calculateNaturalEndpoints was already called, result should have been cached"; assert !called : "calculateNaturalEndpoints was already called, result should have been cached";
called = true; called = true;
return super.calculateNaturalEndpoints(token, metadata, table); return super.calculateNaturalEndpoints(token, metadata);
} }
} }
private AbstractReplicationStrategy getStrategyWithNewTokenMetadata(AbstractReplicationStrategy strategy, TokenMetadata newTmd) throws ConfigurationException
{
return AbstractReplicationStrategy.createReplicationStrategy(
strategy.table,
strategy.getClass().getName(),
newTmd,
strategy.snitch,
strategy.configOptions);
}
} }

4
test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java
View File

@ -188,9 +188,9 @@ public class AntiEntropyServiceTest extends CleanupHelper
addTokens(1 + (2 * DatabaseDescriptor.getReplicationFactor(tablename))); addTokens(1 + (2 * DatabaseDescriptor.getReplicationFactor(tablename)));
AbstractReplicationStrategy ars = StorageService.instance.getReplicationStrategy(tablename); AbstractReplicationStrategy ars = StorageService.instance.getReplicationStrategy(tablename);
Set<InetAddress> expected = new HashSet<InetAddress>(); Set<InetAddress> expected = new HashSet<InetAddress>();
for (Range replicaRange : ars.getAddressRanges(tablename).get(FBUtilities.getLocalAddress())) for (Range replicaRange : ars.getAddressRanges().get(FBUtilities.getLocalAddress()))
{ {
expected.addAll(ars.getRangeAddresses(tmd, tablename).get(replicaRange)); expected.addAll(ars.getRangeAddresses(tmd).get(replicaRange));
} }
expected.remove(FBUtilities.getLocalAddress()); expected.remove(FBUtilities.getLocalAddress());
assertEquals(expected, AntiEntropyService.getNeighbors(tablename)); assertEquals(expected, AntiEntropyService.getNeighbors(tablename));

105
test/unit/org/apache/cassandra/service/MoveTest.java
View File

@ -23,6 +23,7 @@ import java.net.InetAddress;
import java.net.UnknownHostException; import java.net.UnknownHostException;
import java.util.*; import java.util.*;
import org.apache.cassandra.config.ConfigurationException;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*; import static org.junit.Assert.*;
@ -36,7 +37,6 @@ import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.RackUnawareStrategy; import org.apache.cassandra.locator.RackUnawareStrategy;
import org.apache.cassandra.locator.SimpleSnitch; import org.apache.cassandra.locator.SimpleSnitch;
import org.apache.cassandra.locator.TokenMetadata; import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.utils.Pair;
public class MoveTest extends CleanupHelper public class MoveTest extends CleanupHelper
{ {
@ -63,10 +63,8 @@ public class MoveTest extends CleanupHelper
TokenMetadata tmd = ss.getTokenMetadata(); TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe(); tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner(); IPartitioner partitioner = new RandomPartitioner();
AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch());
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategies = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endpointTokens = new ArrayList<Token>(); ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>(); ArrayList<Token> keyTokens = new ArrayList<Token>();
@ -74,52 +72,53 @@ public class MoveTest extends CleanupHelper
createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, RING_SIZE); createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, RING_SIZE);
final Map<Pair<String, Token>, List<InetAddress>> expectedEndpoints = new HashMap<Pair<String, Token>, List<InetAddress>>(); Map<Token, List<InetAddress>> expectedEndpoints = new HashMap<Token, List<InetAddress>>();
for (String table : DatabaseDescriptor.getNonSystemTables()) for (String table : DatabaseDescriptor.getNonSystemTables())
{ {
for (Token token : keyTokens) for (Token token : keyTokens)
{ {
List<InetAddress> endpoints = new ArrayList<InetAddress>(); List<InetAddress> endpoints = new ArrayList<InetAddress>();
Pair<String, Token> key = new Pair<String, Token>(table, token);
Iterator<Token> tokenIter = TokenMetadata.ringIterator(tmd.sortedTokens(), token); Iterator<Token> tokenIter = TokenMetadata.ringIterator(tmd.sortedTokens(), token);
while (tokenIter.hasNext()) while (tokenIter.hasNext())
{ {
endpoints.add(tmd.getEndpoint(tokenIter.next())); endpoints.add(tmd.getEndpoint(tokenIter.next()));
} }
expectedEndpoints.put(key, endpoints); expectedEndpoints.put(token, endpoints);
} }
} }
// Third node leaves // Third node leaves
ss.onChange(hosts.get(LEAVING_NODE), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endpointTokens.get(LEAVING_NODE)))); ss.onChange(hosts.get(LEAVING_NODE),
StorageService.MOVE_STATE,
new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endpointTokens.get(LEAVING_NODE))));
assertTrue(tmd.isLeaving(hosts.get(LEAVING_NODE))); assertTrue(tmd.isLeaving(hosts.get(LEAVING_NODE)));
AbstractReplicationStrategy strategy;
for (String table : DatabaseDescriptor.getNonSystemTables()) for (String table : DatabaseDescriptor.getNonSystemTables())
{ {
strategy = getStrategy(table, tmd);
for (Token token : keyTokens) for (Token token : keyTokens)
{ {
Pair<String, Token> key = new Pair<String, Token>(table, token);
int replicationFactor = DatabaseDescriptor.getReplicationFactor(table); int replicationFactor = DatabaseDescriptor.getReplicationFactor(table);
HashSet<InetAddress> actual = new HashSet<InetAddress>(tmd.getWriteEndpoints(token, table, testStrategy.calculateNaturalEndpoints(token, tmd, table))); HashSet<InetAddress> actual = new HashSet<InetAddress>(tmd.getWriteEndpoints(token, table, strategy.calculateNaturalEndpoints(token, tmd)));
HashSet<InetAddress> expected = new HashSet<InetAddress>(); HashSet<InetAddress> expected = new HashSet<InetAddress>();
for (int i = 0; i < replicationFactor; i++) for (int i = 0; i < replicationFactor; i++)
{ {
expected.add(expectedEndpoints.get(key).get(i)); expected.add(expectedEndpoints.get(token).get(i));
} }
// if the leaving node is in the endpoint list, // if the leaving node is in the endpoint list,
// then we should expect it plus one extra for when it's gone // then we should expect it plus one extra for when it's gone
if (expected.contains(hosts.get(LEAVING_NODE))) if (expected.contains(hosts.get(LEAVING_NODE)))
expected.add(expectedEndpoints.get(key).get(replicationFactor)); expected.add(expectedEndpoints.get(token).get(replicationFactor));
assertEquals("mismatched endpoint sets", expected, actual); assertEquals("mismatched endpoint sets", expected, actual);
} }
} }
ss.setPartitionerUnsafe(oldPartitioner); ss.setPartitionerUnsafe(oldPartitioner);
ss.setReplicationStrategyUnsafe(oldStrategies);
} }
/** /**
@ -127,17 +126,15 @@ public class MoveTest extends CleanupHelper
* simultaneously * simultaneously
*/ */
@Test @Test
public void testSimultaneousMove() throws UnknownHostException public void testSimultaneousMove() throws UnknownHostException, ConfigurationException
{ {
StorageService ss = StorageService.instance; StorageService ss = StorageService.instance;
final int RING_SIZE = 10; final int RING_SIZE = 10;
TokenMetadata tmd = ss.getTokenMetadata(); TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe(); tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner(); IPartitioner partitioner = new RandomPartitioner();
AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch());
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endpointTokens = new ArrayList<Token>(); ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>(); ArrayList<Token> keyTokens = new ArrayList<Token>();
@ -150,7 +147,7 @@ public class MoveTest extends CleanupHelper
final int[] LEAVING = new int[] {6, 8, 9}; final int[] LEAVING = new int[] {6, 8, 9};
for (int leaving : LEAVING) for (int leaving : LEAVING)
ss.onChange(hosts.get(leaving), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endpointTokens.get(leaving)))); ss.onChange(hosts.get(leaving), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endpointTokens.get(leaving))));
// boot two new nodes with keyTokens.get(5) and keyTokens.get(7) // boot two new nodes with keyTokens.get(5) and keyTokens.get(7)
InetAddress boot1 = InetAddress.getByName("127.0.1.1"); InetAddress boot1 = InetAddress.getByName("127.0.1.1");
ss.onChange(boot1, StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_BOOTSTRAPPING + StorageService.Delimiter + partitioner.getTokenFactory().toString(keyTokens.get(5)))); ss.onChange(boot1, StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_BOOTSTRAPPING + StorageService.Delimiter + partitioner.getTokenFactory().toString(keyTokens.get(5))));
@ -160,7 +157,11 @@ public class MoveTest extends CleanupHelper
Collection<InetAddress> endpoints = null; Collection<InetAddress> endpoints = null;
/* don't require test update every time a new keyspace is added to test/conf/cassandra.yaml */ /* don't require test update every time a new keyspace is added to test/conf/cassandra.yaml */
List<String> tables = Arrays.asList("Keyspace1", "Keyspace2", "Keyspace3", "Keyspace4"); Map<String, AbstractReplicationStrategy> tableStrategyMap = new HashMap<String, AbstractReplicationStrategy>();
for (int i=1; i<=4; i++)
{
tableStrategyMap.put("Keyspace" + i, getStrategy("Keyspace" + i, tmd));
}
// pre-calculate the results. // pre-calculate the results.
Map<String, Multimap<Token, InetAddress>> expectedEndpoints = new HashMap<String, Multimap<Token, InetAddress>>(); Map<String, Multimap<Token, InetAddress>> expectedEndpoints = new HashMap<String, Multimap<Token, InetAddress>>();
@ -209,11 +210,14 @@ public class MoveTest extends CleanupHelper
expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3")); expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3"));
expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.1", "127.0.0.2", "127.0.0.3")); expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.1", "127.0.0.2", "127.0.0.3"));
for (String table : tables) for (Map.Entry<String, AbstractReplicationStrategy> tableStrategy : tableStrategyMap.entrySet())
{ {
String table = tableStrategy.getKey();
AbstractReplicationStrategy strategy = tableStrategy.getValue();
for (int i = 0; i < keyTokens.size(); i++) for (int i = 0; i < keyTokens.size(); i++)
{ {
endpoints = tmd.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(i), table, strategy.getNaturalEndpoints(keyTokens.get(i)));
assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).size() == endpoints.size()); assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).size() == endpoints.size());
assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).containsAll(endpoints)); assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).containsAll(endpoints));
} }
@ -224,7 +228,7 @@ public class MoveTest extends CleanupHelper
// tokens 5, 15 and 25 should go three nodes // tokens 5, 15 and 25 should go three nodes
for (int i=0; i<3; ++i) for (int i=0; i<3; ++i)
{ {
endpoints = tmd.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(i), table, strategy.getNaturalEndpoints(keyTokens.get(i)));
assertTrue(endpoints.size() == 3); assertTrue(endpoints.size() == 3);
assertTrue(endpoints.contains(hosts.get(i+1))); assertTrue(endpoints.contains(hosts.get(i+1)));
assertTrue(endpoints.contains(hosts.get(i+2))); assertTrue(endpoints.contains(hosts.get(i+2)));
@ -232,7 +236,7 @@ public class MoveTest extends CleanupHelper
} }
// token 35 should go to nodes 4, 5, 6, 7 and boot1 // token 35 should go to nodes 4, 5, 6, 7 and boot1
endpoints = tmd.getWriteEndpoints(keyTokens.get(3), table, testStrategy.getNaturalEndpoints(keyTokens.get(3), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(3), table, strategy.getNaturalEndpoints(keyTokens.get(3)));
assertTrue(endpoints.size() == 5); assertTrue(endpoints.size() == 5);
assertTrue(endpoints.contains(hosts.get(4))); assertTrue(endpoints.contains(hosts.get(4)));
assertTrue(endpoints.contains(hosts.get(5))); assertTrue(endpoints.contains(hosts.get(5)));
@ -241,7 +245,7 @@ public class MoveTest extends CleanupHelper
assertTrue(endpoints.contains(boot1)); assertTrue(endpoints.contains(boot1));
// token 45 should go to nodes 5, 6, 7, 0, boot1 and boot2 // token 45 should go to nodes 5, 6, 7, 0, boot1 and boot2
endpoints = tmd.getWriteEndpoints(keyTokens.get(4), table, testStrategy.getNaturalEndpoints(keyTokens.get(4), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(4), table, strategy.getNaturalEndpoints(keyTokens.get(4)));
assertTrue(endpoints.size() == 6); assertTrue(endpoints.size() == 6);
assertTrue(endpoints.contains(hosts.get(5))); assertTrue(endpoints.contains(hosts.get(5)));
assertTrue(endpoints.contains(hosts.get(6))); assertTrue(endpoints.contains(hosts.get(6)));
@ -251,7 +255,7 @@ public class MoveTest extends CleanupHelper
assertTrue(endpoints.contains(boot2)); assertTrue(endpoints.contains(boot2));
// token 55 should go to nodes 6, 7, 8, 0, 1, boot1 and boot2 // token 55 should go to nodes 6, 7, 8, 0, 1, boot1 and boot2
endpoints = tmd.getWriteEndpoints(keyTokens.get(5), table, testStrategy.getNaturalEndpoints(keyTokens.get(5), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(5), table, strategy.getNaturalEndpoints(keyTokens.get(5)));
assertTrue(endpoints.size() == 7); assertTrue(endpoints.size() == 7);
assertTrue(endpoints.contains(hosts.get(6))); assertTrue(endpoints.contains(hosts.get(6)));
assertTrue(endpoints.contains(hosts.get(7))); assertTrue(endpoints.contains(hosts.get(7)));
@ -262,7 +266,7 @@ public class MoveTest extends CleanupHelper
assertTrue(endpoints.contains(boot2)); assertTrue(endpoints.contains(boot2));
// token 65 should go to nodes 7, 8, 9, 0, 1 and boot2 // token 65 should go to nodes 7, 8, 9, 0, 1 and boot2
endpoints = tmd.getWriteEndpoints(keyTokens.get(6), table, testStrategy.getNaturalEndpoints(keyTokens.get(6), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(6), table, strategy.getNaturalEndpoints(keyTokens.get(6)));
assertTrue(endpoints.size() == 6); assertTrue(endpoints.size() == 6);
assertTrue(endpoints.contains(hosts.get(7))); assertTrue(endpoints.contains(hosts.get(7)));
assertTrue(endpoints.contains(hosts.get(8))); assertTrue(endpoints.contains(hosts.get(8)));
@ -272,7 +276,7 @@ public class MoveTest extends CleanupHelper
assertTrue(endpoints.contains(boot2)); assertTrue(endpoints.contains(boot2));
// token 75 should to go nodes 8, 9, 0, 1, 2 and boot2 // token 75 should to go nodes 8, 9, 0, 1, 2 and boot2
endpoints = tmd.getWriteEndpoints(keyTokens.get(7), table, testStrategy.getNaturalEndpoints(keyTokens.get(7), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(7), table, strategy.getNaturalEndpoints(keyTokens.get(7)));
assertTrue(endpoints.size() == 6); assertTrue(endpoints.size() == 6);
assertTrue(endpoints.contains(hosts.get(8))); assertTrue(endpoints.contains(hosts.get(8)));
assertTrue(endpoints.contains(hosts.get(9))); assertTrue(endpoints.contains(hosts.get(9)));
@ -282,7 +286,7 @@ public class MoveTest extends CleanupHelper
assertTrue(endpoints.contains(boot2)); assertTrue(endpoints.contains(boot2));
// token 85 should go to nodes 9, 0, 1 and 2 // token 85 should go to nodes 9, 0, 1 and 2
endpoints = tmd.getWriteEndpoints(keyTokens.get(8), table, testStrategy.getNaturalEndpoints(keyTokens.get(8), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(8), table, strategy.getNaturalEndpoints(keyTokens.get(8)));
assertTrue(endpoints.size() == 4); assertTrue(endpoints.size() == 4);
assertTrue(endpoints.contains(hosts.get(9))); assertTrue(endpoints.contains(hosts.get(9)));
assertTrue(endpoints.contains(hosts.get(0))); assertTrue(endpoints.contains(hosts.get(0)));
@ -290,7 +294,7 @@ public class MoveTest extends CleanupHelper
assertTrue(endpoints.contains(hosts.get(2))); assertTrue(endpoints.contains(hosts.get(2)));
// token 95 should go to nodes 0, 1 and 2 // token 95 should go to nodes 0, 1 and 2
endpoints = tmd.getWriteEndpoints(keyTokens.get(9), table, testStrategy.getNaturalEndpoints(keyTokens.get(9), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(9), table, strategy.getNaturalEndpoints(keyTokens.get(9)));
assertTrue(endpoints.size() == 3); assertTrue(endpoints.size() == 3);
assertTrue(endpoints.contains(hosts.get(0))); assertTrue(endpoints.contains(hosts.get(0)));
assertTrue(endpoints.contains(hosts.get(1))); assertTrue(endpoints.contains(hosts.get(1)));
@ -324,11 +328,14 @@ public class MoveTest extends CleanupHelper
expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("75")).removeAll(makeAddrs("127.0.0.10")); expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("75")).removeAll(makeAddrs("127.0.0.10"));
expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("85")).removeAll(makeAddrs("127.0.0.10")); expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("85")).removeAll(makeAddrs("127.0.0.10"));
for (String table : tables) for (Map.Entry<String, AbstractReplicationStrategy> tableStrategy : tableStrategyMap.entrySet())
{ {
String table = tableStrategy.getKey();
AbstractReplicationStrategy strategy = tableStrategy.getValue();
for (int i = 0; i < keyTokens.size(); i++) for (int i = 0; i < keyTokens.size(); i++)
{ {
endpoints = tmd.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(i), table, strategy.getNaturalEndpoints(keyTokens.get(i)));
assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).size() == endpoints.size()); assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).size() == endpoints.size());
assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).containsAll(endpoints)); assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).containsAll(endpoints));
} }
@ -339,7 +346,7 @@ public class MoveTest extends CleanupHelper
// tokens 5, 15 and 25 should go three nodes // tokens 5, 15 and 25 should go three nodes
for (int i=0; i<3; ++i) for (int i=0; i<3; ++i)
{ {
endpoints = tmd.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(i), table, strategy.getNaturalEndpoints(keyTokens.get(i)));
assertTrue(endpoints.size() == 3); assertTrue(endpoints.size() == 3);
assertTrue(endpoints.contains(hosts.get(i+1))); assertTrue(endpoints.contains(hosts.get(i+1)));
assertTrue(endpoints.contains(hosts.get(i+2))); assertTrue(endpoints.contains(hosts.get(i+2)));
@ -347,21 +354,21 @@ public class MoveTest extends CleanupHelper
} }
// token 35 goes to nodes 4, 5 and boot1 // token 35 goes to nodes 4, 5 and boot1
endpoints = tmd.getWriteEndpoints(keyTokens.get(3), table, testStrategy.getNaturalEndpoints(keyTokens.get(3), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(3), table, strategy.getNaturalEndpoints(keyTokens.get(3)));
assertTrue(endpoints.size() == 3); assertTrue(endpoints.size() == 3);
assertTrue(endpoints.contains(hosts.get(4))); assertTrue(endpoints.contains(hosts.get(4)));
assertTrue(endpoints.contains(hosts.get(5))); assertTrue(endpoints.contains(hosts.get(5)));
assertTrue(endpoints.contains(boot1)); assertTrue(endpoints.contains(boot1));
// token 45 goes to nodes 5, boot1 and node7 // token 45 goes to nodes 5, boot1 and node7
endpoints = tmd.getWriteEndpoints(keyTokens.get(4), table, testStrategy.getNaturalEndpoints(keyTokens.get(4), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(4), table, strategy.getNaturalEndpoints(keyTokens.get(4)));
assertTrue(endpoints.size() == 3); assertTrue(endpoints.size() == 3);
assertTrue(endpoints.contains(hosts.get(5))); assertTrue(endpoints.contains(hosts.get(5)));
assertTrue(endpoints.contains(boot1)); assertTrue(endpoints.contains(boot1));
assertTrue(endpoints.contains(hosts.get(7))); assertTrue(endpoints.contains(hosts.get(7)));
// token 55 goes to boot1, 7, boot2, 8 and 0 // token 55 goes to boot1, 7, boot2, 8 and 0
endpoints = tmd.getWriteEndpoints(keyTokens.get(5), table, testStrategy.getNaturalEndpoints(keyTokens.get(5), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(5), table, strategy.getNaturalEndpoints(keyTokens.get(5)));
assertTrue(endpoints.size() == 5); assertTrue(endpoints.size() == 5);
assertTrue(endpoints.contains(boot1)); assertTrue(endpoints.contains(boot1));
assertTrue(endpoints.contains(hosts.get(7))); assertTrue(endpoints.contains(hosts.get(7)));
@ -370,7 +377,7 @@ public class MoveTest extends CleanupHelper
assertTrue(endpoints.contains(hosts.get(0))); assertTrue(endpoints.contains(hosts.get(0)));
// token 65 goes to nodes 7, boot2, 8, 0 and 1 // token 65 goes to nodes 7, boot2, 8, 0 and 1
endpoints = tmd.getWriteEndpoints(keyTokens.get(6), table, testStrategy.getNaturalEndpoints(keyTokens.get(6), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(6), table, strategy.getNaturalEndpoints(keyTokens.get(6)));
assertTrue(endpoints.size() == 5); assertTrue(endpoints.size() == 5);
assertTrue(endpoints.contains(hosts.get(7))); assertTrue(endpoints.contains(hosts.get(7)));
assertTrue(endpoints.contains(boot2)); assertTrue(endpoints.contains(boot2));
@ -379,7 +386,7 @@ public class MoveTest extends CleanupHelper
assertTrue(endpoints.contains(hosts.get(1))); assertTrue(endpoints.contains(hosts.get(1)));
// token 75 goes to nodes boot2, 8, 0, 1 and 2 // token 75 goes to nodes boot2, 8, 0, 1 and 2
endpoints = tmd.getWriteEndpoints(keyTokens.get(7), table, testStrategy.getNaturalEndpoints(keyTokens.get(7), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(7), table, strategy.getNaturalEndpoints(keyTokens.get(7)));
assertTrue(endpoints.size() == 5); assertTrue(endpoints.size() == 5);
assertTrue(endpoints.contains(boot2)); assertTrue(endpoints.contains(boot2));
assertTrue(endpoints.contains(hosts.get(8))); assertTrue(endpoints.contains(hosts.get(8)));
@ -388,14 +395,14 @@ public class MoveTest extends CleanupHelper
assertTrue(endpoints.contains(hosts.get(2))); assertTrue(endpoints.contains(hosts.get(2)));
// token 85 goes to nodes 0, 1 and 2 // token 85 goes to nodes 0, 1 and 2
endpoints = tmd.getWriteEndpoints(keyTokens.get(8), table, testStrategy.getNaturalEndpoints(keyTokens.get(8), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(8), table, strategy.getNaturalEndpoints(keyTokens.get(8)));
assertTrue(endpoints.size() == 3); assertTrue(endpoints.size() == 3);
assertTrue(endpoints.contains(hosts.get(0))); assertTrue(endpoints.contains(hosts.get(0)));
assertTrue(endpoints.contains(hosts.get(1))); assertTrue(endpoints.contains(hosts.get(1)));
assertTrue(endpoints.contains(hosts.get(2))); assertTrue(endpoints.contains(hosts.get(2)));
// token 95 goes to nodes 0, 1 and 2 // token 95 goes to nodes 0, 1 and 2
endpoints = tmd.getWriteEndpoints(keyTokens.get(9), table, testStrategy.getNaturalEndpoints(keyTokens.get(9), table)); endpoints = tmd.getWriteEndpoints(keyTokens.get(9), table, strategy.getNaturalEndpoints(keyTokens.get(9)));
assertTrue(endpoints.size() == 3); assertTrue(endpoints.size() == 3);
assertTrue(endpoints.contains(hosts.get(0))); assertTrue(endpoints.contains(hosts.get(0)));
assertTrue(endpoints.contains(hosts.get(1))); assertTrue(endpoints.contains(hosts.get(1)));
@ -403,7 +410,6 @@ public class MoveTest extends CleanupHelper
} }
ss.setPartitionerUnsafe(oldPartitioner); ss.setPartitionerUnsafe(oldPartitioner);
ss.setReplicationStrategyUnsafe(oldStrategy);
} }
@Test @Test
@ -413,10 +419,8 @@ public class MoveTest extends CleanupHelper
TokenMetadata tmd = ss.getTokenMetadata(); TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe(); tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner(); IPartitioner partitioner = new RandomPartitioner();
AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch());
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endpointTokens = new ArrayList<Token>(); ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>(); ArrayList<Token> keyTokens = new ArrayList<Token>();
@ -472,7 +476,6 @@ public class MoveTest extends CleanupHelper
assertTrue(tmd.getBootstrapTokens().isEmpty()); assertTrue(tmd.getBootstrapTokens().isEmpty());
ss.setPartitionerUnsafe(oldPartitioner); ss.setPartitionerUnsafe(oldPartitioner);
ss.setReplicationStrategyUnsafe(oldStrategy);
} }
@Test @Test
@ -482,10 +485,8 @@ public class MoveTest extends CleanupHelper
TokenMetadata tmd = ss.getTokenMetadata(); TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe(); tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner(); IPartitioner partitioner = new RandomPartitioner();
AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch());
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endpointTokens = new ArrayList<Token>(); ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>(); ArrayList<Token> keyTokens = new ArrayList<Token>();
@ -516,7 +517,6 @@ public class MoveTest extends CleanupHelper
assertTrue(tmd.getToken(hosts.get(2)).equals(keyTokens.get(4))); assertTrue(tmd.getToken(hosts.get(2)).equals(keyTokens.get(4)));
ss.setPartitionerUnsafe(oldPartitioner); ss.setPartitionerUnsafe(oldPartitioner);
ss.setReplicationStrategyUnsafe(oldStrategy);
} }
@Test @Test
@ -526,10 +526,8 @@ public class MoveTest extends CleanupHelper
TokenMetadata tmd = ss.getTokenMetadata(); TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe(); tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner(); IPartitioner partitioner = new RandomPartitioner();
AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch());
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endpointTokens = new ArrayList<Token>(); ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>(); ArrayList<Token> keyTokens = new ArrayList<Token>();
@ -566,7 +564,6 @@ public class MoveTest extends CleanupHelper
assertFalse(tmd.isLeaving(hosts.get(2))); assertFalse(tmd.isLeaving(hosts.get(2)));
ss.setPartitionerUnsafe(oldPartitioner); ss.setPartitionerUnsafe(oldPartitioner);
ss.setReplicationStrategyUnsafe(oldStrategy);
} }
@Test @Test
@ -576,10 +573,8 @@ public class MoveTest extends CleanupHelper
TokenMetadata tmd = ss.getTokenMetadata(); TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe(); tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner(); IPartitioner partitioner = new RandomPartitioner();
AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch());
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endpointTokens = new ArrayList<Token>(); ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>(); ArrayList<Token> keyTokens = new ArrayList<Token>();
@ -608,7 +603,6 @@ public class MoveTest extends CleanupHelper
assertFalse(tmd.isLeaving(hosts.get(2))); assertFalse(tmd.isLeaving(hosts.get(2)));
ss.setPartitionerUnsafe(oldPartitioner); ss.setPartitionerUnsafe(oldPartitioner);
ss.setReplicationStrategyUnsafe(oldStrategy);
} }
/** /**
@ -643,4 +637,15 @@ public class MoveTest extends CleanupHelper
addrs.add(InetAddress.getByName(host)); addrs.add(InetAddress.getByName(host));
return addrs; return addrs;
} }
private AbstractReplicationStrategy getStrategy(String table, TokenMetadata tmd) throws ConfigurationException
{
return AbstractReplicationStrategy.createReplicationStrategy(
table,
"org.apache.cassandra.locator.RackUnawareStrategy",
tmd,
new SimpleSnitch(),
null);
}
} }