From df8a93358c15861e200239fc4e9c23d38f28222f Mon Sep 17 00:00:00 2001 From: Eric Evans Date: Sat, 31 Jul 2010 15:42:27 +0000 Subject: [PATCH] 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 --- conf/cassandra.yaml | 21 ++- interface/cassandra.genavro | 1 + interface/cassandra.thrift | 5 +- .../cassandra/avro/CassandraServer.java | 15 +- .../apache/cassandra/client/RingCache.java | 4 +- .../cassandra/config/DatabaseDescriptor.java | 27 +++- .../apache/cassandra/config/KSMetaData.java | 26 +++- .../apache/cassandra/config/RawKeyspace.java | 3 + .../cassandra/config/ReplicationStrategy.java | 9 ++ .../db/migration/AddColumnFamily.java | 2 +- .../db/migration/DropColumnFamily.java | 2 +- .../db/migration/RenameColumnFamily.java | 2 +- .../db/migration/RenameKeyspace.java | 2 +- .../apache/cassandra/dht/BootStrapper.java | 4 +- .../locator/AbstractReplicationStrategy.java | 131 +++++++++++------- .../locator/DatacenterShardStrategy.java | 65 +++------ .../cassandra/locator/LocalStrategy.java | 7 +- .../cassandra/locator/RackAwareStrategy.java | 13 +- .../locator/RackUnawareStrategy.java | 15 +- .../DatacenterQuorumResponseHandler.java | 2 +- .../DatacenterSyncWriteResponseHandler.java | 14 +- .../DatacenterWriteResponseHandler.java | 4 +- .../cassandra/service/StorageProxy.java | 10 +- .../cassandra/service/StorageService.java | 44 +++--- .../cassandra/thrift/CassandraServer.java | 5 +- test/system/__init__.py | 8 +- test/system/test_thrift_server.py | 2 +- .../config/DatabaseDescriptorTest.java | 4 +- .../org/apache/cassandra/db/DefsTest.java | 6 +- .../locator/DatacenterShardStrategyTest.java | 22 ++- .../locator/RackAwareStrategyTest.java | 8 +- .../locator/RackUnawareStrategyTest.java | 53 +++++-- .../ReplicationStrategyEndpointCacheTest.java | 84 ++++++----- .../service/AntiEntropyServiceTest.java | 4 +- .../apache/cassandra/service/MoveTest.java | 105 +++++++------- 35 files changed, 427 insertions(+), 302 deletions(-) create mode 100644 src/java/org/apache/cassandra/config/ReplicationStrategy.java diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index bc2512e154..1910d41e71 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -185,7 +185,8 @@ request_scheduler_id: keyspace # - name: name of the keyspace; "system" and "definitions" are # reserved for Cassandra Internals. # - 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 # * org.apache.cassandra.locator.RackUnawareStrategy # * org.apache.cassandra.locator.RackAwareStrategy @@ -201,9 +202,17 @@ request_scheduler_id: keyspace # different rack in in the first. # # DatacenterShardStrategy is a generalization of RackAwareStrategy. -# For each datacenter, you can specify (in `datacenter.properties`) -# how many replicas you want on a per-keyspace basis. Replicas are -# placed on different racks within each DC, if possible. +# For each datacenter, you can specify how many replicas you want +# on a per-keyspace basis. Replicas are placed on different racks +# 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 # - 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). # - preload_row_cache: If true, will populate row cache on startup. # Defaults to false. -# - gc_grace_seconds: specifies the time to wait before garbage -# collecting tombstones (deletion markers). defaults to 864000 (10 +# - gc_grace_seconds: specifies the time to wait before garbage +# collecting tombstones (deletion markers). defaults to 864000 (10 # days). See http://wiki.apache.org/cassandra/DistributedDeletes # # NOTE: this keyspace definition is for demonstration purposes only. diff --git a/interface/cassandra.genavro b/interface/cassandra.genavro index b26c4a78e5..a7513d30ca 100644 --- a/interface/cassandra.genavro +++ b/interface/cassandra.genavro @@ -120,6 +120,7 @@ protocol Cassandra { record KsDef { string name; string strategy_class; + union{ map, null } strategy_options; int replication_factor; array cf_defs; } diff --git a/interface/cassandra.thrift b/interface/cassandra.thrift index 814793f13f..2ee350f2d6 100644 --- a/interface/cassandra.thrift +++ b/interface/cassandra.thrift @@ -375,8 +375,9 @@ struct CfDef { struct KsDef { 1: required string name, 2: required string strategy_class, - 3: required i32 replication_factor, - 5: required list cf_defs, + 3: optional map strategy_options, + 4: required i32 replication_factor, + 5: required list cf_defs, } service Cassandra { diff --git a/src/java/org/apache/cassandra/avro/CassandraServer.java b/src/java/org/apache/cassandra/avro/CassandraServer.java index a18bccc349..1e2bbe9d37 100644 --- a/src/java/org/apache/cassandra/avro/CassandraServer.java +++ b/src/java/org/apache/cassandra/avro/CassandraServer.java @@ -661,10 +661,23 @@ public class CassandraServer implements Cassandra { Collections.emptyMap()); cfDefs.add(cfmeta); } - + + + Map strategyOptions = null; + if (ksDef.strategy_options != null && !ksDef.strategy_options.isEmpty()) + { + strategyOptions = new HashMap(); + for (Map.Entry option : ksDef.strategy_options.entrySet()) + { + strategyOptions.put(option.getKey().toString(), option.getValue().toString()); + } + } + + KSMetaData ksmeta = new KSMetaData( ksDef.name.toString(), (Class)Class.forName(ksDef.strategy_class.toString()), + strategyOptions, (int)ksDef.replication_factor, cfDefs.toArray(new CFMetaData[cfDefs.size()])); AddKeyspace add = new AddKeyspace(ksmeta); diff --git a/src/java/org/apache/cassandra/client/RingCache.java b/src/java/org/apache/cassandra/client/RingCache.java index dd4edee38a..08d586f72d 100644 --- a/src/java/org/apache/cassandra/client/RingCache.java +++ b/src/java/org/apache/cassandra/client/RingCache.java @@ -118,7 +118,7 @@ public class RingCache { if (tokenMetadata == null) throw new RuntimeException("Must refresh endpoints before looking up a key."); - AbstractReplicationStrategy strat = StorageService.getReplicationStrategy(tokenMetadata, keyspace); - return strat.getNaturalEndpoints(partitioner_.getToken(key), keyspace); + AbstractReplicationStrategy strat = StorageService.createReplicationStrategy(tokenMetadata, keyspace); + return strat.getNaturalEndpoints(partitioner_.getToken(key)); } } diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 1d6c067b18..a10f3a046d 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -344,11 +344,15 @@ public class DatabaseDescriptor CommitLog.setSegmentSize(conf.commitlog_rotation_threshold_in_mb * 1024 * 1024); // Hardcoded system tables - KSMetaData systemMeta = new KSMetaData(Table.SYSTEM_TABLE, LocalStrategy.class, 1, new CFMetaData[]{CFMetaData.StatusCf, - CFMetaData.HintsCf, - CFMetaData.MigrationsCf, - CFMetaData.SchemaCf, - CFMetaData.StatisticsCf + KSMetaData systemMeta = new KSMetaData(Table.SYSTEM_TABLE, + LocalStrategy.class, + null, + 1, + new CFMetaData[]{CFMetaData.StatusCf, + CFMetaData.HintsCf, + CFMetaData.MigrationsCf, + CFMetaData.SchemaCf, + CFMetaData.StatisticsCf }); CFMetaData.map(CFMetaData.StatusCf); CFMetaData.map(CFMetaData.HintsCf); @@ -618,8 +622,11 @@ public class DatabaseDescriptor cf.gc_grace_seconds, 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; @@ -750,6 +757,12 @@ public class DatabaseDescriptor throw new RuntimeException(table + " not found. Failure to call loadSchemas() perhaps?"); return meta.strategyClass; } + + public static KSMetaData getKSMetaData(String table) + { + assert table != null; + return tables.get(table); + } public static String getJobTrackerAddress() { diff --git a/src/java/org/apache/cassandra/config/KSMetaData.java b/src/java/org/apache/cassandra/config/KSMetaData.java index 4a8dc5aeff..00ba628c30 100644 --- a/src/java/org/apache/cassandra/config/KSMetaData.java +++ b/src/java/org/apache/cassandra/config/KSMetaData.java @@ -36,13 +36,15 @@ public final class KSMetaData { public final String name; public final Class strategyClass; + public final Map strategyOptions; public final int replicationFactor; private final Map cfMetaData; - public KSMetaData(String name, Class strategyClass, int replicationFactor, CFMetaData... cfDefs) + public KSMetaData(String name, Class strategyClass, Map strategyOptions, int replicationFactor, CFMetaData... cfDefs) { this.name = name; this.strategyClass = strategyClass == null ? RackUnawareStrategy.class : strategyClass; + this.strategyOptions = strategyOptions; this.replicationFactor = replicationFactor; Map cfmap = new HashMap(); for (CFMetaData cfm : cfDefs) @@ -59,6 +61,7 @@ public final class KSMetaData KSMetaData other = (KSMetaData)obj; return other.name.equals(name) && ObjectUtils.equals(other.strategyClass, strategyClass) + && ObjectUtils.equals(other.strategyOptions, strategyOptions) && other.replicationFactor == replicationFactor && other.cfMetaData.size() == cfMetaData.size() && other.cfMetaData.equals(cfMetaData); @@ -74,6 +77,14 @@ public final class KSMetaData org.apache.cassandra.avro.KsDef ks = new org.apache.cassandra.avro.KsDef(); ks.name = new Utf8(name); ks.strategy_class = new Utf8(strategyClass.getName()); + if (strategyOptions != null) + { + ks.strategy_options = new HashMap(); + for (Map.Entry e : strategyOptions.entrySet()) + { + ks.strategy_options.put(new Utf8(e.getKey()), new Utf8(e.getValue())); + } + } ks.replication_factor = replicationFactor; ks.cf_defs = SerDeUtils.createArray(cfMetaData.size(), org.apache.cassandra.avro.CfDef.SCHEMA$); 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 { - Class repStratClass = null; + Class repStratClass; try { repStratClass = (Class)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); } + Map strategyOptions = null; + if (ks.strategy_options != null) + { + strategyOptions = new HashMap(); + for (Map.Entry e : ks.strategy_options.entrySet()) + { + strategyOptions.put(e.getKey().toString(), e.getValue().toString()); + } + } int cfsz = (int)ks.cf_defs.size(); CFMetaData[] cfMetaData = new CFMetaData[cfsz]; Iterator cfiter = ks.cf_defs.iterator(); for (int i = 0; i < cfsz; i++) 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); } } diff --git a/src/java/org/apache/cassandra/config/RawKeyspace.java b/src/java/org/apache/cassandra/config/RawKeyspace.java index 2fba2fa930..d800c2ae3c 100644 --- a/src/java/org/apache/cassandra/config/RawKeyspace.java +++ b/src/java/org/apache/cassandra/config/RawKeyspace.java @@ -1,5 +1,7 @@ package org.apache.cassandra.config; +import java.util.Map; + /** * @deprecated Yaml configuration for Keyspaces and ColumnFamilies is deprecated in 0.7 */ @@ -7,6 +9,7 @@ public class RawKeyspace { public String name; public String replica_placement_strategy; + public Map strategy_options; public Integer replication_factor; public RawColumnFamily[] column_families; } diff --git a/src/java/org/apache/cassandra/config/ReplicationStrategy.java b/src/java/org/apache/cassandra/config/ReplicationStrategy.java new file mode 100644 index 0000000000..d094338e9a --- /dev/null +++ b/src/java/org/apache/cassandra/config/ReplicationStrategy.java @@ -0,0 +1,9 @@ +package org.apache.cassandra.config; + +import java.util.Map; + +public class ReplicationStrategy +{ + public String strategy_class; + public Map strategy_options; +} diff --git a/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java b/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java index b30692abcb..b8f9d48a44 100644 --- a/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java +++ b/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java @@ -82,7 +82,7 @@ public class AddColumnFamily extends Migration { List newCfs = new ArrayList(ksm.cfMetaData().values()); 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 diff --git a/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java b/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java index b579405197..6e6055a0f2 100644 --- a/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java +++ b/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java @@ -81,7 +81,7 @@ public class DropColumnFamily extends Migration List newCfs = new ArrayList(ksm.cfMetaData().values()); newCfs.remove(cfm); 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 diff --git a/src/java/org/apache/cassandra/db/migration/RenameColumnFamily.java b/src/java/org/apache/cassandra/db/migration/RenameColumnFamily.java index 66727f46e9..39142637fa 100644 --- a/src/java/org/apache/cassandra/db/migration/RenameColumnFamily.java +++ b/src/java/org/apache/cassandra/db/migration/RenameColumnFamily.java @@ -91,7 +91,7 @@ public class RenameColumnFamily extends Migration assert newCfs.size() == ksm.cfMetaData().size() - 1; CFMetaData newCfm = CFMetaData.rename(oldCfm, newName); 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 diff --git a/src/java/org/apache/cassandra/db/migration/RenameKeyspace.java b/src/java/org/apache/cassandra/db/migration/RenameKeyspace.java index 04b09a44cb..3e0eebef04 100644 --- a/src/java/org/apache/cassandra/db/migration/RenameKeyspace.java +++ b/src/java/org/apache/cassandra/db/migration/RenameKeyspace.java @@ -85,7 +85,7 @@ public class RenameKeyspace extends Migration CFMetaData.purge(oldCf); 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 diff --git a/src/java/org/apache/cassandra/dht/BootStrapper.java b/src/java/org/apache/cassandra/dht/BootStrapper.java index 63327bd419..3f705cce8d 100644 --- a/src/java/org/apache/cassandra/dht/BootStrapper.java +++ b/src/java/org/apache/cassandra/dht/BootStrapper.java @@ -149,10 +149,10 @@ public class BootStrapper { assert tokenMetadata.sortedTokens().size() > 0; final AbstractReplicationStrategy strat = StorageService.instance.getReplicationStrategy(table); - Collection myRanges = strat.getPendingAddressRanges(tokenMetadata, token, address, table); + Collection myRanges = strat.getPendingAddressRanges(tokenMetadata, token, address); Multimap myRangeAddresses = ArrayListMultimap.create(); - Multimap rangeAddresses = strat.getRangeAddresses(tokenMetadata, table); + Multimap rangeAddresses = strat.getRangeAddresses(tokenMetadata); for (Range myRange : myRanges) { for (Range range : rangeAddresses.keySet()) diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index 1aa8415d83..48322d747d 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -19,12 +19,16 @@ 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.util.ArrayList; -import java.util.Collection; -import java.util.Map; -import java.util.Set; +import java.util.*; +import org.apache.cassandra.config.ConfigurationException; +import org.apache.cassandra.service.*; +import org.apache.commons.lang.ObjectUtils; import org.slf4j.Logger; 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.Token; 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.utils.FBUtilities; -import org.apache.cassandra.utils.Pair; import org.cliffc.high_scale_lib.NonBlockingHashMap; /** @@ -50,42 +49,44 @@ public abstract class AbstractReplicationStrategy { private static final Logger logger = LoggerFactory.getLogger(AbstractReplicationStrategy.class); + public String table; private TokenMetadata tokenMetadata; - protected final IEndpointSnitch snitch; - private volatile Map> cachedEndpoints; + public final IEndpointSnitch snitch; + private volatile Map> cachedEndpoints; + public Map configOptions; - AbstractReplicationStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) + AbstractReplicationStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) { + assert table != null; assert snitch != null; assert tokenMetadata != null; this.tokenMetadata = tokenMetadata; this.snitch = snitch; - cachedEndpoints = new NonBlockingHashMap>(); + cachedEndpoints = new NonBlockingHashMap>(); 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), * 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 table the table the natural endpoints are requested for - * @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 endpints + * @return a copy of the natural endpoints for the given token + * @throws IllegalStateException if the number of requested replicas is greater than the number of known endpoints */ - public ArrayList getNaturalEndpoints(Token searchToken, String table) throws IllegalStateException + public ArrayList getNaturalEndpoints(Token searchToken) throws IllegalStateException { - int replicas = getReplicationFactor(table); + int replicas = getReplicationFactor(); Token keyToken = TokenMetadata.firstToken(tokenMetadata.sortedTokens(), searchToken); - EndpointCacheKey cacheKey = new EndpointCacheKey(table, keyToken); - ArrayList endpoints = cachedEndpoints.get(cacheKey); + ArrayList endpoints = cachedEndpoints.get(keyToken); if (endpoints == null) { TokenMetadata tokenMetadataClone = tokenMetadata.cloneOnlyTokenMap(); keyToken = TokenMetadata.firstToken(tokenMetadataClone.sortedTokens(), searchToken); - cacheKey = new EndpointCacheKey(table, keyToken); - endpoints = new ArrayList(calculateNaturalEndpoints(searchToken, tokenMetadataClone, table)); - cachedEndpoints.put(cacheKey, endpoints); + endpoints = new ArrayList(calculateNaturalEndpoints(searchToken, tokenMetadataClone)); + cachedEndpoints.put(keyToken, endpoints); } // 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 table the table the natural endpoints are requested for - * @return a copy of the natural endpoints for the given token and table + * @return a copy of the natural endpoints for the given token * @throws IllegalStateException if the number of requested replicas is greater than the number of known endpoints */ - public abstract Set calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata, String table) throws IllegalStateException; + public abstract Set calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata) throws IllegalStateException; public AbstractWriteResponseHandler getWriteResponseHandler(Collection writeEndpoints, Multimap hintedEndpoints, - ConsistencyLevel consistencyLevel, - String table) + ConsistencyLevel consistencyLevel) { return new WriteResponseHandler(writeEndpoints, hintedEndpoints, consistencyLevel, table); } // instance method so test subclasses can override it - int getReplicationFactor(String table) + int getReplicationFactor() { return DatabaseDescriptor.getReplicationFactor(table); } @@ -170,14 +169,14 @@ public abstract class AbstractReplicationStrategy * (fixing this would probably require merging tokenmetadata into replicationstrategy, * so we could cache/invalidate cleanly.) */ - public Multimap getAddressRanges(TokenMetadata metadata, String table) + public Multimap getAddressRanges(TokenMetadata metadata) { Multimap map = HashMultimap.create(); for (Token token : metadata.sortedTokens()) { Range range = metadata.getPrimaryRangeFor(token); - for (InetAddress ep : calculateNaturalEndpoints(token, metadata, table)) + for (InetAddress ep : calculateNaturalEndpoints(token, metadata)) { map.put(ep, range); } @@ -186,14 +185,14 @@ public abstract class AbstractReplicationStrategy return map; } - public Multimap getRangeAddresses(TokenMetadata metadata, String table) + public Multimap getRangeAddresses(TokenMetadata metadata) { Multimap map = HashMultimap.create(); for (Token token : metadata.sortedTokens()) { Range range = metadata.getPrimaryRangeFor(token); - for (InetAddress ep : calculateNaturalEndpoints(token, metadata, table)) + for (InetAddress ep : calculateNaturalEndpoints(token, metadata)) { map.put(range, ep); } @@ -202,32 +201,27 @@ public abstract class AbstractReplicationStrategy return map; } - public Multimap getAddressRanges(String table) + public Multimap getAddressRanges() { - return getAddressRanges(tokenMetadata, table); + return getAddressRanges(tokenMetadata); } - public Collection getPendingAddressRanges(TokenMetadata metadata, Token pendingToken, InetAddress pendingAddress, String table) + public Collection getPendingAddressRanges(TokenMetadata metadata, Token pendingToken, InetAddress pendingAddress) { TokenMetadata temp = metadata.cloneOnlyTokenMap(); 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); } - protected static class EndpointCacheKey extends Pair - { - public EndpointCacheKey(String table, Token keyToken) {super(table, keyToken);} - } - protected void clearCachedEndpoints() { logger.debug("clearing cached endpoints"); - cachedEndpoints = new NonBlockingHashMap>(); + cachedEndpoints = new NonBlockingHashMap>(); } public void invalidateCachedTokenEndpointValues() @@ -239,4 +233,47 @@ public abstract class AbstractReplicationStrategy { clearCachedEndpoints(); } + + public static AbstractReplicationStrategy createReplicationStrategy(String table, + Class strategyClass, + TokenMetadata tokenMetadata, + IEndpointSnitch snitch, + Map strategyOptions) + throws ConfigurationException + { + AbstractReplicationStrategy strategy; + Class [] parameterTypes = new Class[] {String.class, TokenMetadata.class, IEndpointSnitch.class, Map.class}; + try + { + Constructor 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 strategyOptions) + throws ConfigurationException + { + AbstractReplicationStrategy strategy; + Class c; + try + { + c = (Class) Class.forName(strategyClassName); + } + catch (ClassNotFoundException e) + { + throw new ConfigurationException("Invalid replication strategy class: " + strategyClassName); + } + + return createReplicationStrategy(table, c, tokenMetadata, snitch, strategyOptions); + } } diff --git a/src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java b/src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java index 339702d181..0c9f7e1785 100644 --- a/src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java +++ b/src/java/org/apache/cassandra/locator/DatacenterShardStrategy.java @@ -21,10 +21,12 @@ package org.apache.cassandra.locator; +import java.io.*; import java.net.InetAddress; import java.util.*; import java.util.Map.Entry; +import org.apache.cassandra.config.DatabaseDescriptor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -33,8 +35,6 @@ import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.dht.Token; import org.apache.cassandra.service.*; 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 @@ -51,51 +51,30 @@ import org.apache.cassandra.utils.WrappedRunnable; */ public class DatacenterShardStrategy extends AbstractReplicationStrategy { - private static final String DATACENTER_PROPERTY_FILENAME = "datacenters.properties"; private AbstractRackAwareSnitch snitch; - private volatile Map> datacenters; + private Map datacenters; 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 configOptions) throws ConfigurationException { - super(tokenMetadata, snitch); + super(table, tokenMetadata, snitch, configOptions); if ((!(snitch instanceof AbstractRackAwareSnitch))) throw new IllegalArgumentException("DatacenterShardStrategy requires a rack-aware endpointsnitch"); 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 - { - Properties props = PropertyFileSnitch.resourceToProperties(DATACENTER_PROPERTY_FILENAME); - Map> newDatacenters = new HashMap>(); - for (Entry entry : props.entrySet()) + Map newDatacenters = new HashMap(); + for (Entry entry : configOptions.entrySet()) { - String[] keys = ((String)entry.getKey()).split(":"); - Map map = newDatacenters.get(keys[0]); - if (map == null) - map = new HashMap(); - map.put(keys[1], Integer.parseInt((String) entry.getValue())); - newDatacenters.put(keys[0], map); + newDatacenters.put((String) entry.getKey(), Integer.parseInt((String) entry.getValue())); } + datacenters = Collections.unmodifiableMap(newDatacenters); - logger.info(DATACENTER_PROPERTY_FILENAME + " changed, clearing endpoint cache"); - clearCachedEndpoints(); } - public Set calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata, String table) + public Set calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata) { - int totalReplicas = getReplicationFactor(table); - Map remainingReplicas = new HashMap(datacenters.get(table)); + int totalReplicas = getReplicationFactor(); + Map remainingReplicas = new HashMap(datacenters); Map> dcUsedRacks = new HashMap>(); Set endpoints = new LinkedHashSet(totalReplicas); @@ -152,22 +131,22 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy return endpoints; } - public int getReplicationFactor(String table) + public int getReplicationFactor() { int total = 0; - for (int repFactor : datacenters.get(table).values()) + for (int repFactor : datacenters.values()) total += repFactor; 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 getDatacenters(String table) + public Set 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. */ @Override - public AbstractWriteResponseHandler getWriteResponseHandler(Collection writeEndpoints, Multimap hintedEndpoints, ConsistencyLevel consistency_level, String table) + public AbstractWriteResponseHandler getWriteResponseHandler(Collection writeEndpoints, Multimap hintedEndpoints, ConsistencyLevel consistency_level) { if (consistency_level == ConsistencyLevel.DCQUORUM) { @@ -188,7 +167,7 @@ public class DatacenterShardStrategy extends AbstractReplicationStrategy { 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. */ @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)) { return new DatacenterQuorumResponseHandler(responseResolver, consistencyLevel, table); } - return super.getQuorumResponseHandler(responseResolver, consistencyLevel, table); + return super.getQuorumResponseHandler(responseResolver, consistencyLevel); } } diff --git a/src/java/org/apache/cassandra/locator/LocalStrategy.java b/src/java/org/apache/cassandra/locator/LocalStrategy.java index dc1e6248e2..b05e744cbb 100644 --- a/src/java/org/apache/cassandra/locator/LocalStrategy.java +++ b/src/java/org/apache/cassandra/locator/LocalStrategy.java @@ -21,6 +21,7 @@ package org.apache.cassandra.locator; import java.net.InetAddress; import java.util.HashSet; +import java.util.Map; import java.util.Set; import org.apache.cassandra.utils.FBUtilities; @@ -28,12 +29,12 @@ import org.apache.cassandra.dht.Token; public class LocalStrategy extends AbstractReplicationStrategy { - public LocalStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) + public LocalStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) { - super(tokenMetadata, snitch); + super(table, tokenMetadata, snitch, configOptions); } - public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) + public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata) { Set endpoints = new HashSet(1); InetAddress local = FBUtilities.getLocalAddress(); diff --git a/src/java/org/apache/cassandra/locator/RackAwareStrategy.java b/src/java/org/apache/cassandra/locator/RackAwareStrategy.java index d64adb0baf..0de08d09bb 100644 --- a/src/java/org/apache/cassandra/locator/RackAwareStrategy.java +++ b/src/java/org/apache/cassandra/locator/RackAwareStrategy.java @@ -20,10 +20,7 @@ package org.apache.cassandra.locator; import java.net.InetAddress; -import java.util.ArrayList; -import java.util.LinkedHashSet; -import java.util.Iterator; -import java.util.Set; +import java.util.*; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.Token; @@ -37,16 +34,16 @@ import org.apache.cassandra.dht.Token; */ public class RackAwareStrategy extends AbstractReplicationStrategy { - public RackAwareStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) + public RackAwareStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) { - super(tokenMetadata, snitch); + super(table, tokenMetadata, snitch, configOptions); if (!(snitch instanceof AbstractRackAwareSnitch)) throw new IllegalArgumentException(("RackAwareStrategy requires AbstractRackAwareSnitch.")); } - public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) + public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata) { - int replicas = getReplicationFactor(table); + int replicas = getReplicationFactor(); Set endpoints = new LinkedHashSet(replicas); ArrayList tokens = metadata.sortedTokens(); diff --git a/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java b/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java index ba47f30aa9..bbf5c5668f 100644 --- a/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java +++ b/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java @@ -20,12 +20,8 @@ package org.apache.cassandra.locator; import java.net.InetAddress; -import java.util.ArrayList; -import java.util.LinkedHashSet; -import java.util.Iterator; -import java.util.Set; +import java.util.*; -import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.Token; /** @@ -36,14 +32,14 @@ import org.apache.cassandra.dht.Token; */ public class RackUnawareStrategy extends AbstractReplicationStrategy { - public RackUnawareStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) + public RackUnawareStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) { - super(tokenMetadata, snitch); + super(table, tokenMetadata, snitch, configOptions); } - public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) + public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata) { - int replicas = getReplicationFactor(table); + int replicas = getReplicationFactor(); ArrayList tokens = metadata.sortedTokens(); Set endpoints = new LinkedHashSet(replicas); @@ -62,4 +58,5 @@ public class RackUnawareStrategy extends AbstractReplicationStrategy return endpoints; } + } diff --git a/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java index bbd7edab88..3f6227f77a 100644 --- a/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java @@ -45,6 +45,6 @@ public class DatacenterQuorumResponseHandler extends QuorumResponseHandler public int determineBlockFor(ConsistencyLevel consistency_level, String table) { DatacenterShardStrategy stategy = (DatacenterShardStrategy) StorageService.instance.getReplicationStrategy(table); - return (stategy.getReplicationFactor(localdc, table) / 2) + 1; + return (stategy.getReplicationFactor(localdc) / 2) + 1; } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java index 911395f5b5..5fdecb1e9d 100644 --- a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java @@ -56,7 +56,6 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan private final DatacenterShardStrategy strategy; private HashMap responses = new HashMap(); - private final String table; public DatacenterSyncWriteResponseHandler(Collection writeEndpoints, Multimap hintedEndpoints, ConsistencyLevel consistencyLevel, String table) { @@ -64,12 +63,11 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan super(writeEndpoints, hintedEndpoints, consistencyLevel); assert consistencyLevel == ConsistencyLevel.DCQUORUM; - this.table = 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)); } } @@ -88,14 +86,14 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan return; } - // all the quorum conditionas are met + // all the quorum conditions are met condition.signal(); } public void assureSufficientLiveNodes() throws UnavailableException { Map dcEndpoints = new HashMap(); - for (String dc: strategy.getDatacenters(table)) + for (String dc: strategy.getDatacenters()) dcEndpoints.put(dc, new AtomicInteger()); for (InetAddress destination : hintedEndpoints.keySet()) { @@ -105,8 +103,8 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponseHan dcEndpoints.get(destinationDC).incrementAndGet(); } - // Throw exception if any of the DC doesnt have livenodes to accept write. - for (String dc: strategy.getDatacenters(table)) + // Throw exception if any of the DC doesn't have livenodes to accept write. + for (String dc: strategy.getDatacenters()) { if (dcEndpoints.get(dc).get() != responses.get(dc).get()) throw new UnavailableException(); diff --git a/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java index 021b4b90aa..8a8e1c1d41 100644 --- a/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java @@ -63,7 +63,7 @@ public class DatacenterWriteResponseHandler extends WriteResponseHandler protected int determineBlockFor(String 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(); } } -} \ No newline at end of file +} diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index fcd40f3a1c..0392178eca 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -200,7 +200,7 @@ public class StorageProxy implements StorageProxyMBean Multimap hintedEndpoints = rs.getHintedEndpoints(writeEndpoints); // 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(); 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); } AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.table); - QuorumResponseHandler quorumResponseHandler = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), consistency_level, command.table); + QuorumResponseHandler quorumResponseHandler = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), consistency_level); MessagingService.instance.sendRR(messages, endpoints, quorumResponseHandler); quorumResponseHandlers.add(quorumResponseHandler); commandEndpoints.add(endpoints); @@ -449,7 +449,7 @@ public class StorageProxy implements StorageProxyMBean if (randomlyReadRepair(command)) { AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.table); - QuorumResponseHandler qrhRepair = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), ConsistencyLevel.QUORUM, command.table); + QuorumResponseHandler qrhRepair = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), ConsistencyLevel.QUORUM); if (logger.isDebugEnabled()) logger.debug("Digest mismatch:", ex); Message messageRepair = command.makeReadMessage(); @@ -530,7 +530,7 @@ public class StorageProxy implements StorageProxyMBean // collect replies and resolve according to consistency level RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, liveEndpoints); AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.keyspace); - QuorumResponseHandler> handler = rs.getQuorumResponseHandler(resolver, consistency_level, command.keyspace); + QuorumResponseHandler> handler = rs.getQuorumResponseHandler(resolver, consistency_level); // TODO bail early if live endpoints can't satisfy requested // consistency level for (InetAddress endpoint : liveEndpoints) @@ -795,7 +795,7 @@ public class StorageProxy implements StorageProxyMBean Message message = command.getMessage(); RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace, endpoints); AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy(command.keyspace); - QuorumResponseHandler> handler = rs.getQuorumResponseHandler(resolver, consistency_level, command.keyspace); + QuorumResponseHandler> handler = rs.getQuorumResponseHandler(resolver, consistency_level); MessagingService.instance.sendRR(message, endpoints.get(0), handler); try { diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index b13fb2ff1d..2f7c9324d7 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -21,7 +21,6 @@ package org.apache.cassandra.service; import java.io.IOError; import java.io.IOException; import java.lang.management.ManagementFactory; -import java.lang.reflect.Constructor; import java.net.InetAddress; import java.net.UnknownHostException; 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.util.FileUtils; import org.apache.cassandra.locator.AbstractReplicationStrategy; -import org.apache.cassandra.locator.IEndpointSnitch; import org.apache.cassandra.locator.TokenMetadata; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.ResponseVerbHandler; @@ -274,7 +272,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe public void initReplicationStrategy(String table) { - AbstractReplicationStrategy strat = getReplicationStrategy(tokenMetadata_, table); + AbstractReplicationStrategy strat = createReplicationStrategy(tokenMetadata_, table); replicationStrategies.put(table, strat); } @@ -283,17 +281,19 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe replicationStrategies.remove(table); } - public static AbstractReplicationStrategy getReplicationStrategy(TokenMetadata tokenMetadata, String table) + public static AbstractReplicationStrategy createReplicationStrategy(TokenMetadata tokenMetadata, String table) { - AbstractReplicationStrategy replicationStrategy = null; - Class cls = DatabaseDescriptor.getReplicaPlacementStrategyClass(table); - if (cls == null) - throw new RuntimeException(String.format("No replica strategy configured for %s", table)); - Class [] parameterTypes = new Class[] { TokenMetadata.class, IEndpointSnitch.class}; + AbstractReplicationStrategy replicationStrategy; + KSMetaData ksm = DatabaseDescriptor.getKSMetaData(table); try { - Constructor constructor = cls.getConstructor(parameterTypes); - replicationStrategy = constructor.newInstance(tokenMetadata, DatabaseDescriptor.getEndpointSnitch()); + replicationStrategy = AbstractReplicationStrategy.createReplicationStrategy( + table, + ksm.strategyClass, + tokenMetadata, + DatabaseDescriptor.getEndpointSnitch(), + ksm.strategyOptions + ); } catch (Exception e) { @@ -543,7 +543,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe Map> rangeToEndpointMap = new HashMap>(); for (Range range : ranges) { - rangeToEndpointMap.put(range, getReplicationStrategy(keyspace).getNaturalEndpoints(range.right, keyspace)); + rangeToEndpointMap.put(range, getReplicationStrategy(keyspace).getNaturalEndpoints(range.right)); } return rangeToEndpointMap; } @@ -785,7 +785,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe return; } - Multimap addressRanges = strategy.getAddressRanges(table); + Multimap addressRanges = strategy.getAddressRanges(); // Copy of metadata reflecting the situation after all leave operations are finished. TokenMetadata allLeftMetadata = tm.cloneAfterAllLeft(); @@ -799,8 +799,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe // all leaving nodes are gone. for (Range range : affectedRanges) { - Set currentEndpoints = strategy.calculateNaturalEndpoints(range.right, tm, table); - Set newEndpoints = strategy.calculateNaturalEndpoints(range.right, allLeftMetadata, table); + Set currentEndpoints = strategy.calculateNaturalEndpoints(range.right, tm); + Set newEndpoints = strategy.calculateNaturalEndpoints(range.right, allLeftMetadata); newEndpoints.removeAll(currentEndpoints); pendingRanges.putAll(range, newEndpoints); } @@ -815,7 +815,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe InetAddress endpoint = entry.getValue(); 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); allLeftMetadata.removeEndpoint(endpoint); } @@ -860,7 +860,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe if (logger_.isDebugEnabled()) logger_.debug(endpoint + " was removed, my added ranges: " + StringUtils.join(myNewRanges, ", ")); - Multimap rangeAddresses = getReplicationStrategy(table).getRangeAddresses(tokenMetadata_, table); + Multimap rangeAddresses = getReplicationStrategy(table).getRangeAddresses(tokenMetadata_); Multimap sourceRanges = HashMultimap.create(); 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 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(); @@ -927,7 +927,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe // range. for (Range range : ranges) { - Set newReplicaEndpoints = getReplicationStrategy(table).calculateNaturalEndpoints(range.right, temp, table); + Set newReplicaEndpoints = getReplicationStrategy(table).calculateNaturalEndpoints(range.right, temp); newReplicaEndpoints.removeAll(currentReplicaEndpoints.get(range)); if (logger_.isDebugEnabled()) if (newReplicaEndpoints.isEmpty()) @@ -1227,7 +1227,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe */ Collection 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 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 getLiveNaturalEndpoints(String table, Token token) { List liveEps = new ArrayList(); - List endpoints = getReplicationStrategy(table).getNaturalEndpoints(token, table); + List endpoints = getReplicationStrategy(table).getNaturalEndpoints(token); for (InetAddress endpoint : endpoints) { diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index 33025334d9..36c5542830 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -889,8 +889,9 @@ public class CassandraServer implements Cassandra.Iface KSMetaData ksm = new KSMetaData( ks_def.name, - (Class)Class.forName(ks_def.strategy_class), - ks_def.replication_factor, + (Class)Class.forName(ks_def.strategy_class), + ks_def.strategy_options, + ks_def.replication_factor, cfDefs.toArray(new CFMetaData[cfDefs.size()])); applyMigrationOnStage(new AddKeyspace(ksm)); return DatabaseDescriptor.getDefsVersion().toString(); diff --git a/test/system/__init__.py b/test/system/__init__.py index 251e235060..2d51bc923a 100644 --- a/test/system/__init__.py +++ b/test/system/__init__.py @@ -142,7 +142,7 @@ class ThriftTester(BaseTester): self.client.transport.close() 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', 'Standard2'), @@ -156,7 +156,7 @@ class ThriftTester(BaseTester): 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', 'Standard3'), @@ -164,12 +164,12 @@ class ThriftTester(BaseTester): 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'), ]) - 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', 'Standard3'), diff --git a/test/system/test_thrift_server.py b/test/system/test_thrift_server.py index 770c9664ba..af39898cb0 100644 --- a/test/system/test_thrift_server.py +++ b/test/system/test_thrift_server.py @@ -1124,7 +1124,7 @@ class TestMutations(ThriftTester): def test_system_keyspace_operations(self): """ Test keyspace (add, drop, rename) operations """ # create - keyspace = KsDef('CreateKeyspace', 'org.apache.cassandra.locator.RackUnawareStrategy', 1, + keyspace = KsDef('CreateKeyspace', 'org.apache.cassandra.locator.RackUnawareStrategy', {}, 1, [CfDef('CreateKeyspace', 'CreateKsCf')]) client.system_add_keyspace(keyspace) newks = client.describe_keyspace('CreateKeyspace') diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java index 2b8f27a069..b572eb2389 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java @@ -83,9 +83,9 @@ public class DatabaseDescriptorTest assert DatabaseDescriptor.getNonSystemTables().size() == 0; // 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(); - AddKeyspace ks1 = new AddKeyspace(new KSMetaData("ks1", RackUnawareStrategy.class, 3)); + AddKeyspace ks1 = new AddKeyspace(new KSMetaData("ks1", RackUnawareStrategy.class, null, 3)); ks1.apply(); assert DatabaseDescriptor.getTableDefinition("ks0") != null; diff --git a/test/unit/org/apache/cassandra/db/DefsTest.java b/test/unit/org/apache/cassandra/db/DefsTest.java index b20562f0b7..a5abbe88ce 100644 --- a/test/unit/org/apache/cassandra/db/DefsTest.java +++ b/test/unit/org/apache/cassandra/db/DefsTest.java @@ -66,7 +66,7 @@ public class DefsTest extends CleanupHelper try { 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) { @@ -262,7 +262,7 @@ public class DefsTest extends CleanupHelper { 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.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(); @@ -414,7 +414,7 @@ public class DefsTest extends CleanupHelper { 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(); assert DatabaseDescriptor.getTableDefinition("EmptyKeyspace") != null; diff --git a/test/unit/org/apache/cassandra/locator/DatacenterShardStrategyTest.java b/test/unit/org/apache/cassandra/locator/DatacenterShardStrategyTest.java index d4b6469ced..8d5b75a5f6 100644 --- a/test/unit/org/apache/cassandra/locator/DatacenterShardStrategyTest.java +++ b/test/unit/org/apache/cassandra/locator/DatacenterShardStrategyTest.java @@ -23,7 +23,9 @@ import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.ArrayList; +import java.util.HashMap; import java.util.HashSet; +import java.util.Map; import javax.xml.parsers.ParserConfigurationException; import org.junit.Test; @@ -40,16 +42,22 @@ public class DatacenterShardStrategyTest @Test public void testProperties() throws IOException, ParserConfigurationException, SAXException, ConfigurationException { - PropertyFileSnitch snitch = new PropertyFileSnitch(); + IEndpointSnitch snitch = new PropertyFileSnitch(); TokenMetadata metadata = new TokenMetadata(); createDummyTokens(metadata); - // Set the localhost to the tokenmetadata. Embeded cassandra way? - DatacenterShardStrategy strategy = new DatacenterShardStrategy(metadata, snitch); - assert strategy.getReplicationFactor("DC1", table) == 3; - assert strategy.getReplicationFactor("DC2", table) == 2; - assert strategy.getReplicationFactor("DC3", table) == 1; + + Map configOptions = new HashMap(); + configOptions.put("DC1", "3"); + configOptions.put("DC2", "2"); + 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 - ArrayList endpoints = strategy.getNaturalEndpoints(new StringToken("123"), table); + ArrayList endpoints = strategy.getNaturalEndpoints(new StringToken("123")); assert 6 == endpoints.size(); assert 6 == new HashSet(endpoints).size(); // ensure uniqueness } diff --git a/test/unit/org/apache/cassandra/locator/RackAwareStrategyTest.java b/test/unit/org/apache/cassandra/locator/RackAwareStrategyTest.java index 53a6a38828..4ea30b9608 100644 --- a/test/unit/org/apache/cassandra/locator/RackAwareStrategyTest.java +++ b/test/unit/org/apache/cassandra/locator/RackAwareStrategyTest.java @@ -60,7 +60,7 @@ public class RackAwareStrategyTest { 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("10", "15", "254.0.0.2"); addEndpoint("20", "25", "254.0.0.3"); @@ -85,7 +85,7 @@ public class RackAwareStrategyTest { 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("10", "15", "254.0.0.2"); addEndpoint("20", "25", "254.1.0.3"); @@ -111,7 +111,7 @@ public class RackAwareStrategyTest { 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("10", "15", "254.0.0.2"); addEndpoint("20", "25", "254.0.1.3"); @@ -160,7 +160,7 @@ public class RackAwareStrategyTest { for (Token keyToken : keyTokens) { - List endpoints = strategy.getNaturalEndpoints(keyToken, table); + List endpoints = strategy.getNaturalEndpoints(keyToken); for (int j = 0; j < endpoints.size(); j++) { ArrayList hostsExpected = expectedResults.get(keyToken.toString()); diff --git a/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java b/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java index 89bcd7de3f..d98dd56580 100644 --- a/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java +++ b/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java @@ -26,6 +26,7 @@ import java.util.Collection; import java.util.HashSet; import java.util.List; +import org.apache.cassandra.config.ConfigurationException; import org.junit.Test; import static org.junit.Assert.*; @@ -45,7 +46,7 @@ public class RackUnawareStrategyTest extends SchemaLoader try { 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) { @@ -54,26 +55,21 @@ public class RackUnawareStrategyTest extends SchemaLoader } @Test - public void testBigIntegerEndpoints() throws UnknownHostException + public void testBigIntegerEndpoints() throws UnknownHostException, ConfigurationException { - TokenMetadata tmd = new TokenMetadata(); - AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, new SimpleSnitch()); - List endpointTokens = new ArrayList(); List keyTokens = new ArrayList(); for (int i = 0; i < 5; i++) { endpointTokens.add(new BigIntegerToken(String.valueOf(10 * i))); 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 - public void testStringEndpoints() throws UnknownHostException + public void testStringEndpoints() throws UnknownHostException, ConfigurationException { - TokenMetadata tmd = new TokenMetadata(); IPartitioner partitioner = new OrderPreservingPartitioner(); - AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, new SimpleSnitch()); List endpointTokens = new ArrayList(); List keyTokens = new ArrayList(); @@ -81,15 +77,19 @@ public class RackUnawareStrategyTest extends SchemaLoader endpointTokens.add(new StringToken(String.valueOf((char)('a' + i * 2)))); 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, // 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()) { + tmd = new TokenMetadata(); + strategy = getStrategy(table, tmd); List hosts = new ArrayList(); 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++) { - List endpoints = strategy.getNaturalEndpoints(keyTokens[i], table); + List endpoints = strategy.getNaturalEndpoints(keyTokens[i]); assertEquals(DatabaseDescriptor.getReplicationFactor(table), endpoints.size()); List correctEndpoints = new ArrayList(); for (int j = 0; j < endpoints.size(); j++) @@ -111,13 +111,12 @@ public class RackUnawareStrategyTest extends SchemaLoader } @Test - public void testGetEndpointsDuringBootstrap() throws UnknownHostException + public void testGetEndpointsDuringBootstrap() throws UnknownHostException, ConfigurationException { // the token difference will be RING_SIZE * 2. final int RING_SIZE = 10; TokenMetadata tmd = new TokenMetadata(); TokenMetadata oldTmd = StorageServiceAccessor.setTokenMetadata(tmd); - AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, new SimpleSnitch()); Token[] endpointTokens = 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"); tmd.addBootstrapToken(bsToken, bootstrapEndpoint); + AbstractReplicationStrategy strategy = null; for (String table : DatabaseDescriptor.getNonSystemTables()) { + strategy = getStrategy(table, tmd); + StorageService.calculatePendingRanges(strategy, table); + int replicationFactor = DatabaseDescriptor.getReplicationFactor(table); for (int i = 0; i < keyTokens.length; i++) { - Collection endpoints = tmd.getWriteEndpoints(keyTokens[i], table, strategy.getNaturalEndpoints(keyTokens[i], table)); + Collection endpoints = tmd.getWriteEndpoints(keyTokens[i], table, strategy.getNaturalEndpoints(keyTokens[i])); assertTrue(endpoints.size() >= replicationFactor); for (int j = 0; j < replicationFactor; j++) @@ -167,4 +170,24 @@ public class RackUnawareStrategyTest extends SchemaLoader 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); + } } diff --git a/test/unit/org/apache/cassandra/locator/ReplicationStrategyEndpointCacheTest.java b/test/unit/org/apache/cassandra/locator/ReplicationStrategyEndpointCacheTest.java index d8b4e5fc39..740980c73e 100644 --- a/test/unit/org/apache/cassandra/locator/ReplicationStrategyEndpointCacheTest.java +++ b/test/unit/org/apache/cassandra/locator/ReplicationStrategyEndpointCacheTest.java @@ -21,10 +21,9 @@ package org.apache.cassandra.locator; import java.lang.reflect.Constructor; import java.net.InetAddress; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.Set; +import java.util.*; +import org.apache.cassandra.service.StorageService; import org.apache.commons.lang.StringUtils; import org.junit.Test; @@ -39,12 +38,12 @@ public class ReplicationStrategyEndpointCacheTest extends SchemaLoader private Token searchToken; private AbstractReplicationStrategy strategy; - public void setup(Class stratClass) throws Exception + public void setup(Class stratClass, Map strategyOptions) throws Exception { tmd = new TokenMetadata(); 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(20)), InetAddress.getByName("127.0.0.2")); @@ -59,73 +58,73 @@ public class ReplicationStrategyEndpointCacheTest extends SchemaLoader @Test public void testEndpointsWereCached() throws Exception { - runEndpointsWereCachedTest(FakeRackUnawareStrategy.class); - runEndpointsWereCachedTest(FakeRackAwareStrategy.class); - runEndpointsWereCachedTest(FakeDatacenterShardStrategy.class); + runEndpointsWereCachedTest(FakeRackUnawareStrategy.class, null); + runEndpointsWereCachedTest(FakeRackAwareStrategy.class, null); + runEndpointsWereCachedTest(FakeDatacenterShardStrategy.class, new HashMap()); } - public void runEndpointsWereCachedTest(Class stratClass) throws Exception + public void runEndpointsWereCachedTest(Class stratClass, Map configOptions) throws Exception { - setup(stratClass); - assert strategy.getNaturalEndpoints(searchToken, "Keyspace3").equals(strategy.getNaturalEndpoints(searchToken, "Keyspace3")); + setup(stratClass, configOptions); + assert strategy.getNaturalEndpoints(searchToken).equals(strategy.getNaturalEndpoints(searchToken)); } @Test public void testCacheRespectsTokenChanges() throws Exception { - runCacheRespectsTokenChangesTest(RackUnawareStrategy.class); - runCacheRespectsTokenChangesTest(RackAwareStrategy.class); - runCacheRespectsTokenChangesTest(DatacenterShardStrategy.class); + runCacheRespectsTokenChangesTest(RackUnawareStrategy.class, null); + runCacheRespectsTokenChangesTest(RackAwareStrategy.class, null); + runCacheRespectsTokenChangesTest(DatacenterShardStrategy.class, new HashMap()); } - public void runCacheRespectsTokenChangesTest(Class stratClass) throws Exception + public void runCacheRespectsTokenChangesTest(Class stratClass, Map configOptions) throws Exception { - setup(stratClass); + setup(stratClass, configOptions); ArrayList initial; ArrayList endpoints; - endpoints = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); + endpoints = strategy.getNaturalEndpoints(searchToken); assert endpoints.size() == 5 : StringUtils.join(endpoints, ","); // 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")); - endpoints = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); + endpoints = strategy.getNaturalEndpoints(searchToken); assert endpoints.size() == 5 : StringUtils.join(endpoints, ","); assert !endpoints.equals(initial); // test token removal, newly created token - initial = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); + initial = strategy.getNaturalEndpoints(searchToken); 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.contains(InetAddress.getByName("127.0.0.5")); assert !endpoints.equals(initial); // test token change - initial = strategy.getNaturalEndpoints(searchToken, "Keyspace3"); + initial = strategy.getNaturalEndpoints(searchToken); //move .8 after search token but before other DC3 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.equals(initial); + assert !endpoints.equals(initial); } protected static class FakeRackUnawareStrategy extends RackUnawareStrategy { private boolean called = false; - public FakeRackUnawareStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) + public FakeRackUnawareStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) { - super(tokenMetadata, snitch); + super(table, tokenMetadata, snitch, configOptions); } @Override - public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) + public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata) { assert !called : "calculateNaturalEndpoints was already called, result should have been cached"; 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; - public FakeRackAwareStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) + public FakeRackAwareStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) { - super(tokenMetadata, snitch); + super(table, tokenMetadata, snitch, configOptions); } @Override - public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) + public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata) { assert !called : "calculateNaturalEndpoints was already called, result should have been cached"; 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; - public FakeDatacenterShardStrategy(TokenMetadata tokenMetadata, IEndpointSnitch snitch) throws ConfigurationException + public FakeDatacenterShardStrategy(String table, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) throws ConfigurationException { - super(tokenMetadata, snitch); + super(table, tokenMetadata, snitch, configOptions); } @Override - public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata, String table) + public Set calculateNaturalEndpoints(Token token, TokenMetadata metadata) { assert !called : "calculateNaturalEndpoints was already called, result should have been cached"; 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); + } + } diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java index 31bc96a9c6..b5d5f13ac3 100644 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java +++ b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java @@ -188,9 +188,9 @@ public class AntiEntropyServiceTest extends CleanupHelper addTokens(1 + (2 * DatabaseDescriptor.getReplicationFactor(tablename))); AbstractReplicationStrategy ars = StorageService.instance.getReplicationStrategy(tablename); Set expected = new HashSet(); - 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()); assertEquals(expected, AntiEntropyService.getNeighbors(tablename)); diff --git a/test/unit/org/apache/cassandra/service/MoveTest.java b/test/unit/org/apache/cassandra/service/MoveTest.java index 41c8469c43..a1dd99872d 100644 --- a/test/unit/org/apache/cassandra/service/MoveTest.java +++ b/test/unit/org/apache/cassandra/service/MoveTest.java @@ -23,6 +23,7 @@ import java.net.InetAddress; import java.net.UnknownHostException; import java.util.*; +import org.apache.cassandra.config.ConfigurationException; import org.junit.Test; 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.SimpleSnitch; import org.apache.cassandra.locator.TokenMetadata; -import org.apache.cassandra.utils.Pair; public class MoveTest extends CleanupHelper { @@ -63,10 +63,8 @@ public class MoveTest extends CleanupHelper TokenMetadata tmd = ss.getTokenMetadata(); tmd.clearUnsafe(); IPartitioner partitioner = new RandomPartitioner(); - AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch()); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); - Map oldStrategies = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy)); ArrayList endpointTokens = new ArrayList(); ArrayList keyTokens = new ArrayList(); @@ -74,52 +72,53 @@ public class MoveTest extends CleanupHelper createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, RING_SIZE); - final Map, List> expectedEndpoints = new HashMap, List>(); + Map> expectedEndpoints = new HashMap>(); for (String table : DatabaseDescriptor.getNonSystemTables()) { for (Token token : keyTokens) { List endpoints = new ArrayList(); - Pair key = new Pair(table, token); Iterator tokenIter = TokenMetadata.ringIterator(tmd.sortedTokens(), token); while (tokenIter.hasNext()) { endpoints.add(tmd.getEndpoint(tokenIter.next())); } - expectedEndpoints.put(key, endpoints); + expectedEndpoints.put(token, endpoints); } } // 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))); + AbstractReplicationStrategy strategy; for (String table : DatabaseDescriptor.getNonSystemTables()) { + strategy = getStrategy(table, tmd); for (Token token : keyTokens) { - Pair key = new Pair(table, token); int replicationFactor = DatabaseDescriptor.getReplicationFactor(table); - HashSet actual = new HashSet(tmd.getWriteEndpoints(token, table, testStrategy.calculateNaturalEndpoints(token, tmd, table))); + HashSet actual = new HashSet(tmd.getWriteEndpoints(token, table, strategy.calculateNaturalEndpoints(token, tmd))); HashSet expected = new HashSet(); 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, // then we should expect it plus one extra for when it's gone 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); } } ss.setPartitionerUnsafe(oldPartitioner); - ss.setReplicationStrategyUnsafe(oldStrategies); } /** @@ -127,17 +126,15 @@ public class MoveTest extends CleanupHelper * simultaneously */ @Test - public void testSimultaneousMove() throws UnknownHostException + public void testSimultaneousMove() throws UnknownHostException, ConfigurationException { StorageService ss = StorageService.instance; final int RING_SIZE = 10; TokenMetadata tmd = ss.getTokenMetadata(); tmd.clearUnsafe(); IPartitioner partitioner = new RandomPartitioner(); - AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch()); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); - Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy)); ArrayList endpointTokens = new ArrayList(); ArrayList keyTokens = new ArrayList(); @@ -150,7 +147,7 @@ public class MoveTest extends CleanupHelper final int[] LEAVING = new int[] {6, 8, 9}; 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)))); - + // boot two new nodes with keyTokens.get(5) and keyTokens.get(7) 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)))); @@ -160,7 +157,11 @@ public class MoveTest extends CleanupHelper Collection endpoints = null; /* don't require test update every time a new keyspace is added to test/conf/cassandra.yaml */ - List tables = Arrays.asList("Keyspace1", "Keyspace2", "Keyspace3", "Keyspace4"); + Map tableStrategyMap = new HashMap(); + for (int i=1; i<=4; i++) + { + tableStrategyMap.put("Keyspace" + i, getStrategy("Keyspace" + i, tmd)); + } // pre-calculate the results. Map> expectedEndpoints = new HashMap>(); @@ -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("95"), makeAddrs("127.0.0.1", "127.0.0.2", "127.0.0.3")); - for (String table : tables) + for (Map.Entry tableStrategy : tableStrategyMap.entrySet()) { + String table = tableStrategy.getKey(); + AbstractReplicationStrategy strategy = tableStrategy.getValue(); + 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)).containsAll(endpoints)); } @@ -224,7 +228,7 @@ public class MoveTest extends CleanupHelper // tokens 5, 15 and 25 should go three nodes 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.contains(hosts.get(i+1))); 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 - 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.contains(hosts.get(4))); assertTrue(endpoints.contains(hosts.get(5))); @@ -241,7 +245,7 @@ public class MoveTest extends CleanupHelper assertTrue(endpoints.contains(boot1)); // 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.contains(hosts.get(5))); assertTrue(endpoints.contains(hosts.get(6))); @@ -251,7 +255,7 @@ public class MoveTest extends CleanupHelper assertTrue(endpoints.contains(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.contains(hosts.get(6))); assertTrue(endpoints.contains(hosts.get(7))); @@ -262,7 +266,7 @@ public class MoveTest extends CleanupHelper assertTrue(endpoints.contains(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.contains(hosts.get(7))); assertTrue(endpoints.contains(hosts.get(8))); @@ -272,7 +276,7 @@ public class MoveTest extends CleanupHelper assertTrue(endpoints.contains(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.contains(hosts.get(8))); assertTrue(endpoints.contains(hosts.get(9))); @@ -282,7 +286,7 @@ public class MoveTest extends CleanupHelper assertTrue(endpoints.contains(boot2)); // 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.contains(hosts.get(9))); assertTrue(endpoints.contains(hosts.get(0))); @@ -290,7 +294,7 @@ public class MoveTest extends CleanupHelper assertTrue(endpoints.contains(hosts.get(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.contains(hosts.get(0))); 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("85")).removeAll(makeAddrs("127.0.0.10")); - for (String table : tables) + for (Map.Entry tableStrategy : tableStrategyMap.entrySet()) { + String table = tableStrategy.getKey(); + AbstractReplicationStrategy strategy = tableStrategy.getValue(); + 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)).containsAll(endpoints)); } @@ -339,7 +346,7 @@ public class MoveTest extends CleanupHelper // tokens 5, 15 and 25 should go three nodes 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.contains(hosts.get(i+1))); 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 - 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.contains(hosts.get(4))); assertTrue(endpoints.contains(hosts.get(5))); assertTrue(endpoints.contains(boot1)); // 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.contains(hosts.get(5))); assertTrue(endpoints.contains(boot1)); assertTrue(endpoints.contains(hosts.get(7))); // 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.contains(boot1)); assertTrue(endpoints.contains(hosts.get(7))); @@ -370,7 +377,7 @@ public class MoveTest extends CleanupHelper assertTrue(endpoints.contains(hosts.get(0))); // 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.contains(hosts.get(7))); assertTrue(endpoints.contains(boot2)); @@ -379,7 +386,7 @@ public class MoveTest extends CleanupHelper assertTrue(endpoints.contains(hosts.get(1))); // 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.contains(boot2)); assertTrue(endpoints.contains(hosts.get(8))); @@ -388,14 +395,14 @@ public class MoveTest extends CleanupHelper assertTrue(endpoints.contains(hosts.get(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.contains(hosts.get(0))); assertTrue(endpoints.contains(hosts.get(1))); assertTrue(endpoints.contains(hosts.get(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.contains(hosts.get(0))); assertTrue(endpoints.contains(hosts.get(1))); @@ -403,7 +410,6 @@ public class MoveTest extends CleanupHelper } ss.setPartitionerUnsafe(oldPartitioner); - ss.setReplicationStrategyUnsafe(oldStrategy); } @Test @@ -413,10 +419,8 @@ public class MoveTest extends CleanupHelper TokenMetadata tmd = ss.getTokenMetadata(); tmd.clearUnsafe(); IPartitioner partitioner = new RandomPartitioner(); - AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch()); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); - Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy)); ArrayList endpointTokens = new ArrayList(); ArrayList keyTokens = new ArrayList(); @@ -472,7 +476,6 @@ public class MoveTest extends CleanupHelper assertTrue(tmd.getBootstrapTokens().isEmpty()); ss.setPartitionerUnsafe(oldPartitioner); - ss.setReplicationStrategyUnsafe(oldStrategy); } @Test @@ -482,10 +485,8 @@ public class MoveTest extends CleanupHelper TokenMetadata tmd = ss.getTokenMetadata(); tmd.clearUnsafe(); IPartitioner partitioner = new RandomPartitioner(); - AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch()); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); - Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy)); ArrayList endpointTokens = new ArrayList(); ArrayList keyTokens = new ArrayList(); @@ -516,7 +517,6 @@ public class MoveTest extends CleanupHelper assertTrue(tmd.getToken(hosts.get(2)).equals(keyTokens.get(4))); ss.setPartitionerUnsafe(oldPartitioner); - ss.setReplicationStrategyUnsafe(oldStrategy); } @Test @@ -526,10 +526,8 @@ public class MoveTest extends CleanupHelper TokenMetadata tmd = ss.getTokenMetadata(); tmd.clearUnsafe(); IPartitioner partitioner = new RandomPartitioner(); - AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch()); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); - Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy)); ArrayList endpointTokens = new ArrayList(); ArrayList keyTokens = new ArrayList(); @@ -566,7 +564,6 @@ public class MoveTest extends CleanupHelper assertFalse(tmd.isLeaving(hosts.get(2))); ss.setPartitionerUnsafe(oldPartitioner); - ss.setReplicationStrategyUnsafe(oldStrategy); } @Test @@ -576,10 +573,8 @@ public class MoveTest extends CleanupHelper TokenMetadata tmd = ss.getTokenMetadata(); tmd.clearUnsafe(); IPartitioner partitioner = new RandomPartitioner(); - AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, new SimpleSnitch()); IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner); - Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy)); ArrayList endpointTokens = new ArrayList(); ArrayList keyTokens = new ArrayList(); @@ -608,7 +603,6 @@ public class MoveTest extends CleanupHelper assertFalse(tmd.isLeaving(hosts.get(2))); ss.setPartitionerUnsafe(oldPartitioner); - ss.setReplicationStrategyUnsafe(oldStrategy); } /** @@ -643,4 +637,15 @@ public class MoveTest extends CleanupHelper addrs.add(InetAddress.getByName(host)); return addrs; } + + private AbstractReplicationStrategy getStrategy(String table, TokenMetadata tmd) throws ConfigurationException + { + return AbstractReplicationStrategy.createReplicationStrategy( + table, + "org.apache.cassandra.locator.RackUnawareStrategy", + tmd, + new SimpleSnitch(), + null); + } + }