diff --git a/CHANGES.txt b/CHANGES.txt
index a7a93056c6..177cc03fa7 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -19,6 +19,7 @@ dev
* report latency and cache hit rate statistics with lifetime totals
instead of average over the last minute (CASSANDRA-702)
* support get_range_slice for RandomPartitioner (CASSANDRA-745)
+ * per-keyspace replication factory and replication strategy (CASSANDRA-620)
0.5.1
diff --git a/conf/storage-conf.xml b/conf/storage-conf.xml
index 01a3f6a921..324babbc63 100644
--- a/conf/storage-conf.xml
+++ b/conf/storage-conf.xml
@@ -111,6 +111,30 @@
RowsCached="1000"
KeysCachedFraction="0"
Comment="A column family with supercolumns, whose column and subcolumn names are UTF8 strings"/>
+
+
+ org.apache.cassandra.locator.RackUnawareStrategy
+
+
+ 1
+
+
+ org.apache.cassandra.locator.EndPointSnitch
+
@@ -156,29 +180,6 @@
-->
-
- org.apache.cassandra.locator.EndPointSnitch
-
-
- org.apache.cassandra.locator.RackUnawareStrategy
-
-
- 1
-
");
- for ( Range range : ranges_ )
- {
- sb.append(range);
- sb.append(" ");
- }
- return sb.toString();
- }
-
- private static class StreamRequestMetadataSerializer implements ICompactSerializer
- {
- public void serialize(StreamRequestMetadata srMetadata, DataOutputStream dos) throws IOException
- {
- CompactEndPointSerializationHelper.serialize(srMetadata.target_, dos);
- dos.writeInt(srMetadata.ranges_.size());
- for (Range range : srMetadata.ranges_)
- {
- Range.serializer().serialize(range, dos);
- }
- }
-
- public StreamRequestMetadata deserialize(DataInputStream dis) throws IOException
- {
- InetAddress target = CompactEndPointSerializationHelper.deserialize(dis);
- int size = dis.readInt();
- List ranges = (size == 0) ? null : new ArrayList();
- for( int i = 0; i < size; ++i )
- {
- ranges.add(Range.serializer().deserialize(dis));
- }
- return new StreamRequestMetadata( target, ranges );
- }
- }
-}
+package org.apache.cassandra.streaming;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.net.CompactEndPointSerializationHelper;
+
+/**
+ * This encapsulates information of the list of ranges that a target
+ * node requires to be transferred. This will be bundled in a
+ * StreamRequestsMessage and sent to nodes that are going to handoff
+ * the data.
+*/
+class StreamRequestMetadata
+{
+ private static ICompactSerializer serializer_;
+ static
+ {
+ serializer_ = new StreamRequestMetadataSerializer();
+ }
+
+ protected static ICompactSerializer serializer()
+ {
+ return serializer_;
+ }
+
+ protected InetAddress target_;
+ protected Collection ranges_;
+ protected String table_;
+
+ StreamRequestMetadata(InetAddress target, Collection ranges, String table)
+ {
+ target_ = target;
+ ranges_ = ranges;
+ table_ = table;
+ }
+
+ public String toString()
+ {
+ StringBuilder sb = new StringBuilder("");
+ sb.append(table_);
+ sb.append("@");
+ sb.append(target_);
+ sb.append("------->");
+ for ( Range range : ranges_ )
+ {
+ sb.append(range);
+ sb.append(" ");
+ }
+ return sb.toString();
+ }
+}
+
+class StreamRequestMetadataSerializer implements ICompactSerializer
+{
+ public void serialize(StreamRequestMetadata srMetadata, DataOutputStream dos) throws IOException
+ {
+ CompactEndPointSerializationHelper.serialize(srMetadata.target_, dos);
+ dos.writeUTF(srMetadata.table_);
+ dos.writeInt(srMetadata.ranges_.size());
+ for (Range range : srMetadata.ranges_)
+ {
+ Range.serializer().serialize(range, dos);
+ }
+ }
+
+ public StreamRequestMetadata deserialize(DataInputStream dis) throws IOException
+ {
+ InetAddress target = CompactEndPointSerializationHelper.deserialize(dis);
+ String table = dis.readUTF();
+ int size = dis.readInt();
+ List ranges = (size == 0) ? null : new ArrayList();
+ for( int i = 0; i < size; ++i )
+ {
+ ranges.add(Range.serializer().deserialize(dis));
+ }
+ return new StreamRequestMetadata(target, ranges, table);
+ }
+}
diff --git a/src/java/org/apache/cassandra/streaming/StreamRequestVerbHandler.java b/src/java/org/apache/cassandra/streaming/StreamRequestVerbHandler.java
index 97c41428b4..d0d98143b3 100644
--- a/src/java/org/apache/cassandra/streaming/StreamRequestVerbHandler.java
+++ b/src/java/org/apache/cassandra/streaming/StreamRequestVerbHandler.java
@@ -52,7 +52,7 @@ public class StreamRequestVerbHandler implements IVerbHandler
{
if (logger_.isDebugEnabled())
logger_.debug(srm.toString());
- StreamOut.transferRanges(srm.target_, srm.ranges_, null);
+ StreamOut.transferRanges(srm.target_, srm.table_, srm.ranges_, null);
}
}
catch (IOException ex)
diff --git a/src/java/org/apache/cassandra/tools/ClusterCmd.java b/src/java/org/apache/cassandra/tools/ClusterCmd.java
index b2a2113ad2..c55726472e 100644
--- a/src/java/org/apache/cassandra/tools/ClusterCmd.java
+++ b/src/java/org/apache/cassandra/tools/ClusterCmd.java
@@ -144,9 +144,9 @@ public class ClusterCmd {
hf.printHelp(usage, "", options, header);
}
- public void printEndPoints(String key)
+ public void printEndPoints(String key, String table)
{
- List endpoints = probe.getEndPoints(key);
+ List endpoints = probe.getEndPoints(key, table);
System.out.println(String.format("%-17s: %s", "Key", key));
System.out.println(String.format("%-17s: %s", "Endpoints", endpoints));
}
@@ -254,21 +254,21 @@ public class ClusterCmd {
String cmdName = arguments[0];
if (cmdName.equals("get_endpoints"))
{
- if (arguments.length <= 1)
+ if (arguments.length <= 2)
{
- System.err.println("missing key argument");
+ System.err.println("missing key and/or table argument");
}
- clusterCmd.printEndPoints(arguments[1]);
+ clusterCmd.printEndPoints(arguments[1], arguments[2]);
}
- else if (cmdName.equals("global_snapshot"))
- {
+ else if (cmdName.equals("global_snapshot"))
+ {
String snapshotName = "";
if (arguments.length > 1)
{
snapshotName = arguments[1];
}
- clusterCmd.takeGlobalSnapshot(snapshotName);
- }
+ clusterCmd.takeGlobalSnapshot(snapshotName);
+ }
else if (cmdName.equals("clear_global_snapshot"))
{
clusterCmd.clearGlobalSnapshot();
diff --git a/src/java/org/apache/cassandra/tools/NodeCmd.java b/src/java/org/apache/cassandra/tools/NodeCmd.java
index 32257fc3e2..3de909a8a4 100644
--- a/src/java/org/apache/cassandra/tools/NodeCmd.java
+++ b/src/java/org/apache/cassandra/tools/NodeCmd.java
@@ -66,7 +66,7 @@ public class NodeCmd {
*/
public void printRing(PrintStream outs)
{
- Map> rangeMap = probe.getRangeToEndPointMap();
+ Map> rangeMap = probe.getRangeToEndPointMap(null);
List ranges = new ArrayList(rangeMap.keySet());
Collections.sort(ranges);
Set liveNodes = probe.getLiveNodes();
diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java
index 589b1ee83d..d2a5e12b10 100644
--- a/src/java/org/apache/cassandra/tools/NodeProbe.java
+++ b/src/java/org/apache/cassandra/tools/NodeProbe.java
@@ -26,6 +26,8 @@ import java.lang.management.MemoryUsage;
import java.lang.management.RuntimeMXBean;
import java.net.InetAddress;
import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -139,15 +141,84 @@ public class NodeProbe
ssProxy.forceTableRepair(tableName, columnFamilies);
}
- public Map> getRangeToEndPointMap()
+ public Map> getRangeToEndPointMap(String tableName)
{
- return ssProxy.getRangeToEndPointMap();
+ return ssProxy.getRangeToEndPointMap(tableName);
}
public Set getLiveNodes()
{
return ssProxy.getLiveNodes();
}
+
+ /**
+ * Write a textual representation of the Cassandra ring.
+ *
+ * @param outs the stream to write to
+ */
+ public void printRing(PrintStream outs)
+ {
+ Map> rangeMap = ssProxy.getRangeToEndPointMap(null);
+ List ranges = new ArrayList(rangeMap.keySet());
+ Collections.sort(ranges);
+ Set liveNodes = ssProxy.getLiveNodes();
+ Set deadNodes = ssProxy.getUnreachableNodes();
+ Map loadMap = ssProxy.getLoadMap();
+
+ // Print range-to-endpoint mapping
+ int counter = 0;
+ outs.print(String.format("%-14s", "Address"));
+ outs.print(String.format("%-11s", "Status"));
+ outs.print(String.format("%-14s", "Load"));
+ outs.print(String.format("%-43s", "Range"));
+ outs.println("Ring");
+ // emphasize that we're showing the right part of each range
+ if (ranges.size() > 1)
+ {
+ outs.println(String.format("%-14s%-11s%-14s%-43s", "", "", "", ranges.get(0).left));
+ }
+ // normal range & node info
+ for (Range range : ranges) {
+ List endpoints = rangeMap.get(range);
+ String primaryEndpoint = endpoints.get(0);
+
+ outs.print(String.format("%-14s", primaryEndpoint));
+
+ String status = liveNodes.contains(primaryEndpoint)
+ ? "Up"
+ : deadNodes.contains(primaryEndpoint)
+ ? "Down"
+ : "?";
+ outs.print(String.format("%-11s", status));
+
+ String load = loadMap.containsKey(primaryEndpoint) ? loadMap.get(primaryEndpoint) : "?";
+ outs.print(String.format("%-14s", load));
+
+ outs.print(String.format("%-43s", range.right));
+
+ String asciiRingArt;
+ if (counter == 0)
+ {
+ asciiRingArt = "|<--|";
+ }
+ else if (counter == (rangeMap.size() - 1))
+ {
+ asciiRingArt = "|-->|";
+ }
+ else
+ {
+ if ((rangeMap.size() > 4) && ((counter % 2) == 0))
+ asciiRingArt = "v |";
+ else if ((rangeMap.size() > 4) && ((counter % 2) != 0))
+ asciiRingArt = "| ^";
+ else
+ asciiRingArt = "| |";
+ }
+ outs.println(asciiRingArt);
+
+ counter++;
+ }
+ }
public Set getUnreachableNodes()
{
@@ -325,9 +396,9 @@ public class NodeProbe
}
}
- public List getEndPoints(String key)
+ public List getEndPoints(String key, String table)
{
- return ssProxy.getNaturalEndpoints(key);
+ return ssProxy.getNaturalEndpoints(key, table);
}
}
diff --git a/test/conf/storage-conf.xml b/test/conf/storage-conf.xml
index 7ba0e127c7..72fa9e39a2 100644
--- a/test/conf/storage-conf.xml
+++ b/test/conf/storage-conf.xml
@@ -23,9 +23,6 @@
batch
1.0
org.apache.cassandra.dht.CollatingOrderPreservingPartitioner
- org.apache.cassandra.locator.EndPointSnitch
- org.apache.cassandra.locator.RackUnawareStrategy
- 1
5000
127.0.0.1
7010
@@ -51,12 +48,33 @@
+ org.apache.cassandra.locator.RackUnawareStrategy
+ 1
+ org.apache.cassandra.locator.EndPointSnitch
+ org.apache.cassandra.locator.RackUnawareStrategy
+ 1
+ org.apache.cassandra.locator.EndPointSnitch
+
+
+
+ org.apache.cassandra.locator.RackUnawareStrategy
+ 5
+ org.apache.cassandra.locator.EndPointSnitch
+
+
+
+
+
+
+ org.apache.cassandra.locator.RackUnawareStrategy
+ 3
+ org.apache.cassandra.locator.EndPointSnitch
diff --git a/test/system/test_server.py b/test/system/test_server.py
index 4b2e0fd572..0d3249d13f 100644
--- a/test/system/test_server.py
+++ b/test/system/test_server.py
@@ -873,7 +873,7 @@ class TestMutations(CassandraTester):
def test_describe_keyspace(self):
""" Test keyspace description """
kspaces = client.get_string_list_property("keyspaces")
- assert len(kspaces) == 3, kspaces
+ assert len(kspaces) == 5, kspaces # ['Keyspace1', 'Keyspace2', 'Keyspace3', 'Keyspace4', 'system']
ks1 = client.describe_keyspace("Keyspace1")
assert set(ks1.keys()) == set(['Super1', 'Standard1', 'Standard2', 'StandardLong1', 'StandardLong2', 'Super3', 'Super2', 'Super4'])
sysks = client.describe_keyspace("system")
diff --git a/test/unit/org/apache/cassandra/client/TestRingCache.java b/test/unit/org/apache/cassandra/client/TestRingCache.java
index 4cf51c16a2..20c6d2d5ee 100644
--- a/test/unit/org/apache/cassandra/client/TestRingCache.java
+++ b/test/unit/org/apache/cassandra/client/TestRingCache.java
@@ -59,30 +59,51 @@ public class TestRingCache
}
/**
- * usage: java -Dstorage-config="confpath" org.apache.cassandra.client.TestRingCache
+ * usage: java -Dstorage-config="confpath" org.apache.cassandra.client.TestRingCache [keyspace row-id-prefix row-id-int]
+ * to test a single keyspace/row, use the parameters. row-id-prefix and row-id-int are appended together to form a
+ * single row id. If you supply now parameters, 'Keyspace1' is assumed and will check 9 rows ('row1' through 'row9').
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Throwable
{
- String table = "Keyspace1";
- for (int nRows=1; nRows<10; nRows++)
+ String table;
+ int minRow;
+ int maxRow;
+ String rowPrefix;
+ if (args.length > 0)
{
- String row = "row" + nRows;
+ table = args[0];
+ rowPrefix = args[1];
+ minRow = Integer.parseInt(args[2]);
+ maxRow = minRow + 1;
+ }
+ else
+ {
+ table = "Keyspace1";
+ minRow = 1;
+ maxRow = 10;
+ rowPrefix = "row";
+ }
+
+ for (int nRows = minRow; nRows < maxRow; nRows++)
+ {
+ String row = rowPrefix + nRows;
ColumnPath col = createColumnPath("Standard1", null, "col1".getBytes());
- List endPoints = ringCache.getEndPoint(row);
+ List endPoints = ringCache.getEndPoint(table, row);
String hosts="";
for (int i = 0; i < endPoints.size(); i++)
hosts = hosts + ((i > 0) ? "," : "") + endPoints.get(i);
System.out.println("hosts with key " + row + " : " + hosts + "; choose " + endPoints.get(0));
-
+
// now, read the row back directly from the host owning the row locally
setup(endPoints.get(0).getHostAddress(), DatabaseDescriptor.getThriftPort());
thriftClient.insert(table, row, col, "val1".getBytes(), 1, ConsistencyLevel.ONE);
Column column=thriftClient.get(table, row, col, ConsistencyLevel.ONE).column;
System.out.println("read row " + row + " " + new String(column.name) + ":" + new String(column.value) + ":" + column.timestamp);
}
+
System.exit(1);
}
}
diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java
index 42ad5e1c8d..e7ba4c74f4 100644
--- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java
+++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java
@@ -34,8 +34,8 @@ public class DatabaseDescriptorTest
* TODO: A more general method of property modification would be useful, but
* will probably have to wait for a refactor away from all the static fields.
*/
- public static void setReplicationFactor(int factor)
+ public static void setReplicationFactor(String table, int factor)
{
- DatabaseDescriptor.setReplicationFactorUnsafe(factor);
+ DatabaseDescriptor.setReplicationFactorUnsafe(table, factor);
}
}
diff --git a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
index ac9d965878..5639f241e0 100644
--- a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
+++ b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java
@@ -25,6 +25,7 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.commons.lang.StringUtils;
import static org.junit.Assert.assertEquals;
import org.junit.Test;
@@ -71,12 +72,17 @@ public class BootStrapperTest
@Test
public void testSourceTargetComputation() throws UnknownHostException
{
- testSourceTargetComputation(1);
- testSourceTargetComputation(3);
- testSourceTargetComputation(100);
+ final int[] clusterSizes = new int[] { 1, 3, 5, 10, 100};
+ for (String table : DatabaseDescriptor.getNonSystemTables())
+ {
+ int replicationFactor = DatabaseDescriptor.getReplicationFactor(table);
+ for (int clusterSize : clusterSizes)
+ if (clusterSize >= replicationFactor)
+ testSourceTargetComputation(table, clusterSize, replicationFactor);
+ }
}
- private void testSourceTargetComputation(int numOldNodes) throws UnknownHostException
+ private void testSourceTargetComputation(String table, int numOldNodes, int replicationFactor) throws UnknownHostException
{
StorageService ss = StorageService.instance;
@@ -86,8 +92,8 @@ public class BootStrapperTest
TokenMetadata tmd = ss.getTokenMetadata();
assertEquals(numOldNodes, tmd.sortedTokens().size());
- BootStrapper b = new BootStrapper(ss.getReplicationStrategy(), myEndpoint, myToken, tmd);
- Multimap res = b.getRangesWithSources();
+ BootStrapper b = new BootStrapper(myEndpoint, myToken, tmd);
+ Multimap res = b.getRangesWithSources(table);
int transferCount = 0;
for (Map.Entry> e : res.asMap().entrySet())
@@ -96,8 +102,7 @@ public class BootStrapperTest
transferCount++;
}
- /* Only 1 transfer from old node to new node */
- assertEquals(1, transferCount);
+ assertEquals(replicationFactor, transferCount);
IFailureDetector mockFailureDetector = new IFailureDetector()
{
public boolean isAlive(InetAddress ep)
@@ -112,8 +117,10 @@ public class BootStrapperTest
public void remove(InetAddress ep) { throw new UnsupportedOperationException(); }
};
Multimap temp = BootStrapper.getWorkMap(res, mockFailureDetector);
- assertEquals(1, temp.keySet().size());
- assertEquals(1, temp.asMap().values().iterator().next().size());
+ // there isn't any point in testing the size of these collections for any specific size. When a random partitioner
+ // is used, they will vary.
+ assert temp.keySet().size() > 0;
+ assert temp.asMap().values().iterator().next().size() > 0;
assert !temp.keySet().iterator().next().equals(myEndpoint);
}
diff --git a/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java b/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java
index db4c8f22c2..973e76213e 100644
--- a/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java
+++ b/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java
@@ -25,6 +25,9 @@ import java.util.List;
import java.util.ArrayList;
import java.util.Collection;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.gms.ApplicationState;
+import org.apache.cassandra.service.StorageServiceAccessor;
import org.junit.Test;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.RandomPartitioner;
@@ -39,11 +42,27 @@ import java.net.UnknownHostException;
public class RackUnawareStrategyTest
{
+ @Test
+ public void tryBogusTable()
+ {
+ AbstractReplicationStrategy rs = StorageService.instance.getReplicationStrategy("Keyspace1");
+ assertNotNull(rs);
+ try
+ {
+ rs = StorageService.instance.getReplicationStrategy("SomeBogusTableThatDoesntExist");
+ throw new AssertionError("SS.getReplicationStrategy() should have thrown a RuntimeException.");
+ }
+ catch (RuntimeException ex)
+ {
+ // This exception should be thrown.
+ }
+ }
+
@Test
public void testBigIntegerEndpoints() throws UnknownHostException
{
TokenMetadata tmd = new TokenMetadata();
- AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, 3);
+ AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, null);
List endPointTokens = new ArrayList();
List keyTokens = new ArrayList();
@@ -51,7 +70,8 @@ public class RackUnawareStrategyTest
endPointTokens.add(new BigIntegerToken(String.valueOf(10 * i)));
keyTokens.add(new BigIntegerToken(String.valueOf(10 * i + 5)));
}
- testGetEndpoints(tmd, strategy, endPointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0]));
+ for (String table : DatabaseDescriptor.getNonSystemTables())
+ testGetEndpoints(tmd, strategy, endPointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0]), table);
}
@Test
@@ -59,7 +79,7 @@ public class RackUnawareStrategyTest
{
TokenMetadata tmd = new TokenMetadata();
IPartitioner partitioner = new OrderPreservingPartitioner();
- AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, 3);
+ AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, null);
List endPointTokens = new ArrayList();
List keyTokens = new ArrayList();
@@ -67,12 +87,13 @@ public class RackUnawareStrategyTest
endPointTokens.add(new StringToken(String.valueOf((char)('a' + i * 2))));
keyTokens.add(partitioner.getToken(String.valueOf((char)('a' + i * 2 + 1))));
}
- testGetEndpoints(tmd, strategy, endPointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0]));
+ for (String table : DatabaseDescriptor.getNonSystemTables())
+ testGetEndpoints(tmd, strategy, endPointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0]), table);
}
// 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 testGetEndpoints(TokenMetadata tmd, AbstractReplicationStrategy strategy, Token[] endPointTokens, Token[] keyTokens) throws UnknownHostException
+ private void testGetEndpoints(TokenMetadata tmd, AbstractReplicationStrategy strategy, Token[] endPointTokens, Token[] keyTokens, String table) throws UnknownHostException
{
List hosts = new ArrayList();
for (int i = 0; i < endPointTokens.length; i++)
@@ -84,8 +105,8 @@ public class RackUnawareStrategyTest
for (int i = 0; i < keyTokens.length; i++)
{
- List endPoints = strategy.getNaturalEndpoints(keyTokens[i]);
- assertEquals(3, endPoints.size());
+ List endPoints = strategy.getNaturalEndpoints(keyTokens[i], table);
+ assertEquals(DatabaseDescriptor.getReplicationFactor(table), endPoints.size());
for (int j = 0; j < endPoints.size(); j++)
{
assertEquals(endPoints.get(j), hosts.get((i + j + 1) % hosts.size()));
@@ -96,16 +117,19 @@ public class RackUnawareStrategyTest
@Test
public void testGetEndpointsDuringBootstrap() throws UnknownHostException
{
+ // the token difference will be RING_SIZE * 2.
+ final int RING_SIZE = 10;
TokenMetadata tmd = new TokenMetadata();
- AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, 3);
+ TokenMetadata oldTmd = StorageServiceAccessor.setTokenMetadata(tmd);
+ AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, null);
- Token[] endPointTokens = new Token[5];
- Token[] keyTokens = new Token[5];
+ Token[] endPointTokens = new Token[RING_SIZE];
+ Token[] keyTokens = new Token[RING_SIZE];
- for (int i = 0; i < 5; i++)
+ for (int i = 0; i < RING_SIZE; i++)
{
- endPointTokens[i] = new BigIntegerToken(String.valueOf(10 * i));
- keyTokens[i] = new BigIntegerToken(String.valueOf(10 * i + 5));
+ endPointTokens[i] = new BigIntegerToken(String.valueOf(RING_SIZE * 2 * i));
+ keyTokens[i] = new BigIntegerToken(String.valueOf(RING_SIZE * 2 * i + RING_SIZE));
}
List hosts = new ArrayList();
@@ -115,28 +139,36 @@ public class RackUnawareStrategyTest
tmd.updateNormalToken(endPointTokens[i], ep);
hosts.add(ep);
}
-
- //Add bootstrap node id=6
- Token bsToken = new BigIntegerToken(String.valueOf(25));
- InetAddress bootstrapEndPoint = InetAddress.getByName("127.0.0.6");
+
+ // bootstrap at the end of the ring
+ Token bsToken = new BigIntegerToken(String.valueOf(210));
+ InetAddress bootstrapEndPoint = InetAddress.getByName("127.0.0.11");
tmd.addBootstrapToken(bsToken, bootstrapEndPoint);
- StorageService.calculatePendingRanges(tmd, strategy);
- for (int i = 0; i < keyTokens.length; i++)
+ for (String table : DatabaseDescriptor.getNonSystemTables())
{
- Collection endPoints = strategy.getWriteEndpoints(keyTokens[i], strategy.getNaturalEndpoints(keyTokens[i]));
- assertTrue(endPoints.size() >= 3);
+ StorageService.calculatePendingRanges(strategy, table);
+ int replicationFactor = DatabaseDescriptor.getReplicationFactor(table);
- for (int j = 0; j < 3; j++)
+ for (int i = 0; i < keyTokens.length; i++)
{
- //Check that the old nodes are definitely included
- assertTrue(endPoints.contains(hosts.get((i + j + 1) % hosts.size())));
+ Collection endPoints = strategy.getWriteEndpoints(keyTokens[i], table, strategy.getNaturalEndpoints(keyTokens[i], table));
+ assertTrue(endPoints.size() >= replicationFactor);
+
+ for (int j = 0; j < replicationFactor; j++)
+ {
+ //Check that the old nodes are definitely included
+ assertTrue(endPoints.contains(hosts.get((i + j + 1) % hosts.size())));
+ }
+
+ // bootstrapEndPoint should be in the endPoints for i in MAX-RF to MAX, but not in any earlier ep.
+ if (i < RING_SIZE - replicationFactor)
+ assertFalse(endPoints.contains(bootstrapEndPoint));
+ else
+ assertTrue(endPoints.contains(bootstrapEndPoint));
}
- // for 5, 15, 25 this should include bootstrap node
- if (i < 3)
- assertTrue(endPoints.contains(bootstrapEndPoint));
- else
- assertFalse(endPoints.contains(bootstrapEndPoint));
}
+
+ StorageServiceAccessor.setTokenMetadata(oldTmd);
}
}
diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java
index 0c1bb0f0f3..5120f4e9cb 100644
--- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java
+++ b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java
@@ -61,8 +61,9 @@ public class AntiEntropyServiceTest extends CleanupHelper
if (!initialized)
{
LOCAL = FBUtilities.getLocalAddress();
+ tablename = DatabaseDescriptor.getTables().iterator().next();
// bump the replication factor so that local overlaps with REMOTE below
- DatabaseDescriptorTest.setReplicationFactor(2);
+ DatabaseDescriptorTest.setReplicationFactor(tablename, 2);
StorageService.instance.initServer();
// generate a fake endpoint for which we can spoof receiving/sending trees
@@ -72,7 +73,6 @@ public class AntiEntropyServiceTest extends CleanupHelper
tmd.updateNormalToken(part.getMinimumToken(), REMOTE);
assert tmd.isMember(REMOTE);
- tablename = DatabaseDescriptor.getTables().iterator().next();
cfname = Table.open(tablename).getColumnFamilies().iterator().next();
initialized = true;
}
@@ -89,7 +89,7 @@ public class AntiEntropyServiceTest extends CleanupHelper
@Test
public void testGetValidator() throws Throwable
{
- aes.clearNaturalRepairs();
+ aes.clearNaturalRepairs_TestsOnly();
// not major
assert aes.getValidator(tablename, cfname, null, false) instanceof NoopValidator;
@@ -174,13 +174,13 @@ public class AntiEntropyServiceTest extends CleanupHelper
Util.writeColumnFamily(rms);
ColumnFamilyStore store = Util.writeColumnFamily(rms);
- TreePair old = aes.getRendezvousPair(tablename, cfname, REMOTE);
+ TreePair old = aes.getRendezvousPair_TestsOnly(tablename, cfname, REMOTE);
// force a readonly compaction, and wait for it to finish
CompactionManager.instance.submitReadonly(store, REMOTE).get(5000, TimeUnit.MILLISECONDS);
// check that a tree was created and stored
flushAES().get(5000, TimeUnit.MILLISECONDS);
- assert old != aes.getRendezvousPair(tablename, cfname, REMOTE);
+ assert old != aes.getRendezvousPair_TestsOnly(tablename, cfname, REMOTE);
}
@Test
@@ -200,7 +200,7 @@ public class AntiEntropyServiceTest extends CleanupHelper
// confirm that our reference is not equal to the original due
// to (de)serialization
- assert tree != aes.getRendezvousPair(tablename, cfname, REMOTE).left;
+ assert tree != aes.getRendezvousPair_TestsOnly(tablename, cfname, REMOTE).left;
}
@Test
diff --git a/test/unit/org/apache/cassandra/service/MoveTest.java b/test/unit/org/apache/cassandra/service/MoveTest.java
index 8bdddd1d4e..dae67f0a15 100644
--- a/test/unit/org/apache/cassandra/service/MoveTest.java
+++ b/test/unit/org/apache/cassandra/service/MoveTest.java
@@ -24,8 +24,13 @@ import java.util.*;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.commons.lang.StringUtils;
import org.junit.Test;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import org.apache.cassandra.dht.IPartitioner;
@@ -40,6 +45,16 @@ import org.apache.cassandra.gms.ApplicationState;
public class MoveTest
{
+ // handy way of creating a mapping of strategies to use in StorageService.
+ private static Map createReplacements(AbstractReplicationStrategy strat)
+ {
+ Map replacements = new HashMap();
+ for (String table : DatabaseDescriptor.getTables())
+ replacements.put(table, strat);
+ return replacements;
+ }
+
+
/**
* Test whether write endpoints is correct when the node is leaving. Uses
* StorageService.onChange and does not manipulate token metadata directly.
@@ -48,53 +63,74 @@ public class MoveTest
public void testWriteEndPointsDuringLeave() throws UnknownHostException
{
StorageService ss = StorageService.instance;
+ final int RING_SIZE = 5;
+ final int LEAVING_NODE = 2;
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
- AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, 3);
+ AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, null);
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
- AbstractReplicationStrategy oldStrategy = ss.setReplicationStrategyUnsafe(testStrategy);
+ Map oldStrategies = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList endPointTokens = new ArrayList();
ArrayList keyTokens = new ArrayList();
List hosts = new ArrayList();
- createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, 5);
+ createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, RING_SIZE);
+ final Map> deadNodesRanges = new HashMap>();
+ for (String table : DatabaseDescriptor.getNonSystemTables())
+ {
+ List list = new ArrayList();
+ list.addAll(testStrategy.getAddressRanges(table).get(hosts.get(LEAVING_NODE)));
+ Collections.sort(list);
+ deadNodesRanges.put(table, list);
+ }
+
// Third node leaves
- ss.onChange(hosts.get(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(2))));
+ ss.onChange(hosts.get(LEAVING_NODE), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(LEAVING_NODE))));
// check that it is correctly marked as leaving in tmd
- assertTrue(tmd.isLeaving(hosts.get(2)));
+ assertTrue(tmd.isLeaving(hosts.get(LEAVING_NODE)));
// check that pending ranges are correct (primary range should go to 1st node, first
// replica range to 4th node and 2nd replica range to 5th node)
- assertTrue(tmd.getPendingRanges(hosts.get(0)).get(0).equals(new Range(endPointTokens.get(1),
- endPointTokens.get(2))));
- assertTrue(tmd.getPendingRanges(hosts.get(3)).get(0).equals(new Range(endPointTokens.get(4),
- endPointTokens.get(0))));
- assertTrue(tmd.getPendingRanges(hosts.get(4)).get(0).equals(new Range(endPointTokens.get(0),
- endPointTokens.get(1))));
-
- for (int i=0; i endPoints = testStrategy.getWriteEndpoints(keyTokens.get(i), testStrategy.getNaturalEndpoints(keyTokens.get(i)));
+ int replicationFactor = DatabaseDescriptor.getReplicationFactor(table);
- // Original third node does not store replicas for 4th and 5th node (ranges 20-30
- // and 30-40 respectively), so their write endpoints count should be still 3. The
- // third node stores data for ranges 40-0, 0-10 and 10-20, so writes falling to
- // these ranges should have four endpoints now. keyTokens[2] is 25 and keyTokens[3]
- // is 35, so these are the ones that should have 3 endpoints.
- if (i==2 || i==3)
- assertTrue(endPoints.size() == 3);
- else
- assertTrue(endPoints.size() == 4);
+ // if the ring minus the leaving node leaves us with less than RF, we're hosed.
+ if (hosts.size()-1 < replicationFactor)
+ continue;
+
+ // verify that the replicationFactor nodes after the leaving node are gatherings it's pending ranges.
+ // in the case where rf==5, we're screwed because we basically just lost data.
+ for (int i = 0; i < replicationFactor; i++)
+ {
+ assertTrue(tmd.getPendingRanges(table, hosts.get((LEAVING_NODE + 1 + i) % RING_SIZE)).size() > 0);
+ assertEquals(tmd.getPendingRanges(table, hosts.get((LEAVING_NODE + 1 + i) % RING_SIZE)).get(0), deadNodesRanges.get(table).get(i));
+ }
+
+ // note that we're iterating over nodes and sample tokens.
+ final int replicaStart = (LEAVING_NODE-replicationFactor+RING_SIZE)%RING_SIZE;
+ for (int i=0; i endPoints = testStrategy.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table));
+ // figure out if this node is one of the nodes previous to the failed node (2).
+ boolean isReplica = (i - replicaStart + RING_SIZE) % RING_SIZE < replicationFactor;
+ // the preceeding leaving_node-replication_factor nodes should have and additional ep (replication_factor+1);
+ if (isReplica)
+ assertTrue(endPoints.size() == replicationFactor + 1);
+ else
+ assertTrue(endPoints.size() == replicationFactor);
+
+ }
}
ss.setPartitionerUnsafe(oldPartitioner);
- ss.setReplicationStrategyUnsafe(oldStrategy);
+ ss.setReplicationStrategyUnsafe(oldStrategies);
}
/**
@@ -105,25 +141,26 @@ public class MoveTest
public void testSimultaneousMove() throws UnknownHostException
{
StorageService ss = StorageService.instance;
+ final int RING_SIZE = 10;
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
- AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, 3);
+ AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, null);
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
- AbstractReplicationStrategy oldStrategy = ss.setReplicationStrategyUnsafe(testStrategy);
+ Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList endPointTokens = new ArrayList();
ArrayList keyTokens = new ArrayList();
List hosts = new ArrayList();
// create a ring or 10 nodes
- createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, 10);
+ createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, RING_SIZE);
// nodes 6, 8 and 9 leave
- ss.onChange(hosts.get(6), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(6))));
- ss.onChange(hosts.get(8), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(8))));
- ss.onChange(hosts.get(9), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(9))));
+ 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");
@@ -133,152 +170,246 @@ public class MoveTest
Collection endPoints = null;
- // tokens 5, 15 and 25 should go three nodes
- for (int i=0; i<3; ++i)
+ // pre-calculate the results.
+ Map> expectedEndpoints = new HashMap>();
+ expectedEndpoints.put("Keyspace1", HashMultimap.create());
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2"));
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3"));
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4"));
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("35"), makeAddrs("127.0.0.5"));
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("45"), makeAddrs("127.0.0.6"));
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("55"), makeAddrs("127.0.0.7", "127.0.0.8", "127.0.1.1"));
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("65"), makeAddrs("127.0.0.8"));
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.1.2", "127.0.0.1"));
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.10", "127.0.0.1"));
+ expectedEndpoints.get("Keyspace1").putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.1"));
+ expectedEndpoints.put("Keyspace2", HashMultimap.create());
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2"));
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3"));
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4"));
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("35"), makeAddrs("127.0.0.5"));
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("45"), makeAddrs("127.0.0.6"));
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("55"), makeAddrs("127.0.0.7", "127.0.0.8", "127.0.1.1"));
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("65"), makeAddrs("127.0.0.8"));
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.1.2", "127.0.0.1"));
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.10", "127.0.0.1"));
+ expectedEndpoints.get("Keyspace2").putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.1"));
+ expectedEndpoints.put("Keyspace3", HashMultimap.create());
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5", "127.0.0.6"));
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3", "127.0.0.4", "127.0.0.5", "127.0.0.6", "127.0.0.7", "127.0.1.1", "127.0.0.8"));
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4", "127.0.0.5", "127.0.0.6", "127.0.0.7", "127.0.0.8", "127.0.1.2", "127.0.0.1", "127.0.1.1"));
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("35"), makeAddrs("127.0.0.5", "127.0.0.6", "127.0.0.7", "127.0.0.8", "127.0.0.9", "127.0.1.2", "127.0.0.1", "127.0.0.2", "127.0.1.1"));
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("45"), makeAddrs("127.0.0.6", "127.0.0.7", "127.0.0.8", "127.0.0.9", "127.0.0.10", "127.0.1.2", "127.0.0.1", "127.0.0.2", "127.0.1.1", "127.0.0.3"));
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("55"), makeAddrs("127.0.0.7", "127.0.0.8", "127.0.0.9", "127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.1.1", "127.0.1.2"));
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("65"), makeAddrs("127.0.0.8", "127.0.0.9", "127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.1.2", "127.0.0.3", "127.0.0.4"));
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.1.2", "127.0.0.4", "127.0.0.5"));
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5"));
+ expectedEndpoints.get("Keyspace3").putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5"));
+ expectedEndpoints.put("Keyspace4", HashMultimap.create());
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("5"), makeAddrs("127.0.0.2", "127.0.0.3", "127.0.0.4"));
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("15"), makeAddrs("127.0.0.3", "127.0.0.4", "127.0.0.5"));
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("25"), makeAddrs("127.0.0.4", "127.0.0.5", "127.0.0.6"));
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("35"), makeAddrs("127.0.0.5", "127.0.0.6", "127.0.0.7", "127.0.1.1", "127.0.0.8"));
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("45"), makeAddrs("127.0.0.6", "127.0.0.7", "127.0.0.8", "127.0.1.2", "127.0.0.1", "127.0.1.1"));
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("55"), makeAddrs("127.0.0.7", "127.0.0.8", "127.0.0.9", "127.0.0.1", "127.0.0.2", "127.0.1.1", "127.0.1.2"));
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("65"), makeAddrs("127.0.0.8", "127.0.0.9", "127.0.0.10", "127.0.1.2", "127.0.0.1", "127.0.0.2"));
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("75"), makeAddrs("127.0.0.9", "127.0.0.10", "127.0.0.1", "127.0.1.2", "127.0.0.2", "127.0.0.3"));
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("85"), makeAddrs("127.0.0.10", "127.0.0.1", "127.0.0.2", "127.0.0.3"));
+ expectedEndpoints.get("Keyspace4").putAll(new BigIntegerToken("95"), makeAddrs("127.0.0.1", "127.0.0.2", "127.0.0.3"));
+
+ for (String table : DatabaseDescriptor.getNonSystemTables())
{
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(i), testStrategy.getNaturalEndpoints(keyTokens.get(i)));
+ for (int i = 0; i < keyTokens.size(); i++)
+ {
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table));
+ assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).size() == endPoints.size());
+ assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).containsAll(endPoints));
+ }
+
+ // just to be sure that things still work according to the old tests, run them:
+ if (DatabaseDescriptor.getReplicationFactor(table) != 3)
+ continue;
+ // tokens 5, 15 and 25 should go three nodes
+ for (int i=0; i<3; ++i)
+ {
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table));
+ assertTrue(endPoints.size() == 3);
+ assertTrue(endPoints.contains(hosts.get(i+1)));
+ assertTrue(endPoints.contains(hosts.get(i+2)));
+ assertTrue(endPoints.contains(hosts.get(i+3)));
+ }
+
+ // token 35 should go to nodes 4, 5, 6, 7 and boot1
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(3), table, testStrategy.getNaturalEndpoints(keyTokens.get(3), table));
+ assertTrue(endPoints.size() == 5);
+ assertTrue(endPoints.contains(hosts.get(4)));
+ assertTrue(endPoints.contains(hosts.get(5)));
+ assertTrue(endPoints.contains(hosts.get(6)));
+ assertTrue(endPoints.contains(hosts.get(7)));
+ assertTrue(endPoints.contains(boot1));
+
+ // token 45 should go to nodes 5, 6, 7, 0, boot1 and boot2
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(4), table, testStrategy.getNaturalEndpoints(keyTokens.get(4), table));
+ assertTrue(endPoints.size() == 6);
+ assertTrue(endPoints.contains(hosts.get(5)));
+ assertTrue(endPoints.contains(hosts.get(6)));
+ assertTrue(endPoints.contains(hosts.get(7)));
+ assertTrue(endPoints.contains(hosts.get(0)));
+ assertTrue(endPoints.contains(boot1));
+ assertTrue(endPoints.contains(boot2));
+
+ // token 55 should go to nodes 6, 7, 8, 0, 1, boot1 and boot2
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(5), table, testStrategy.getNaturalEndpoints(keyTokens.get(5), table));
+ assertTrue(endPoints.size() == 7);
+ assertTrue(endPoints.contains(hosts.get(6)));
+ assertTrue(endPoints.contains(hosts.get(7)));
+ assertTrue(endPoints.contains(hosts.get(8)));
+ assertTrue(endPoints.contains(hosts.get(0)));
+ assertTrue(endPoints.contains(hosts.get(1)));
+ assertTrue(endPoints.contains(boot1));
+ assertTrue(endPoints.contains(boot2));
+
+ // token 65 should go to nodes 7, 8, 9, 0, 1 and boot2
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(6), table, testStrategy.getNaturalEndpoints(keyTokens.get(6), table));
+ assertTrue(endPoints.size() == 6);
+ assertTrue(endPoints.contains(hosts.get(7)));
+ assertTrue(endPoints.contains(hosts.get(8)));
+ assertTrue(endPoints.contains(hosts.get(9)));
+ assertTrue(endPoints.contains(hosts.get(0)));
+ assertTrue(endPoints.contains(hosts.get(1)));
+ assertTrue(endPoints.contains(boot2));
+
+ // token 75 should to go nodes 8, 9, 0, 1, 2 and boot2
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(7), table, testStrategy.getNaturalEndpoints(keyTokens.get(7), table));
+ assertTrue(endPoints.size() == 6);
+ assertTrue(endPoints.contains(hosts.get(8)));
+ assertTrue(endPoints.contains(hosts.get(9)));
+ assertTrue(endPoints.contains(hosts.get(0)));
+ assertTrue(endPoints.contains(hosts.get(1)));
+ assertTrue(endPoints.contains(hosts.get(2)));
+ assertTrue(endPoints.contains(boot2));
+
+ // token 85 should go to nodes 9, 0, 1 and 2
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(8), table, testStrategy.getNaturalEndpoints(keyTokens.get(8), table));
+ assertTrue(endPoints.size() == 4);
+ assertTrue(endPoints.contains(hosts.get(9)));
+ assertTrue(endPoints.contains(hosts.get(0)));
+ assertTrue(endPoints.contains(hosts.get(1)));
+ assertTrue(endPoints.contains(hosts.get(2)));
+
+ // token 95 should go to nodes 0, 1 and 2
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(9), table, testStrategy.getNaturalEndpoints(keyTokens.get(9), table));
assertTrue(endPoints.size() == 3);
- assertTrue(endPoints.contains(hosts.get(i+1)));
- assertTrue(endPoints.contains(hosts.get(i+2)));
- assertTrue(endPoints.contains(hosts.get(i+3)));
+ assertTrue(endPoints.contains(hosts.get(0)));
+ assertTrue(endPoints.contains(hosts.get(1)));
+ assertTrue(endPoints.contains(hosts.get(2)));
+
}
- // token 35 should go to nodes 4, 5, 6, 7 and boot1
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(3), testStrategy.getNaturalEndpoints(keyTokens.get(3)));
- assertTrue(endPoints.size() == 5);
- assertTrue(endPoints.contains(hosts.get(4)));
- assertTrue(endPoints.contains(hosts.get(5)));
- assertTrue(endPoints.contains(hosts.get(6)));
- assertTrue(endPoints.contains(hosts.get(7)));
- assertTrue(endPoints.contains(boot1));
-
- // token 45 should go to nodes 5, 6, 7, 0, boot1 and boot2
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(4), testStrategy.getNaturalEndpoints(keyTokens.get(4)));
- assertTrue(endPoints.size() == 6);
- assertTrue(endPoints.contains(hosts.get(5)));
- assertTrue(endPoints.contains(hosts.get(6)));
- assertTrue(endPoints.contains(hosts.get(7)));
- assertTrue(endPoints.contains(hosts.get(0)));
- assertTrue(endPoints.contains(boot1));
- assertTrue(endPoints.contains(boot2));
-
- // token 55 should go to nodes 6, 7, 8, 0, 1, boot1 and boot2
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(5), testStrategy.getNaturalEndpoints(keyTokens.get(5)));
- assertTrue(endPoints.size() == 7);
- assertTrue(endPoints.contains(hosts.get(6)));
- assertTrue(endPoints.contains(hosts.get(7)));
- assertTrue(endPoints.contains(hosts.get(8)));
- assertTrue(endPoints.contains(hosts.get(0)));
- assertTrue(endPoints.contains(hosts.get(1)));
- assertTrue(endPoints.contains(boot1));
- assertTrue(endPoints.contains(boot2));
-
- // token 65 should go to nodes 7, 8, 9, 0, 1 and boot2
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(6), testStrategy.getNaturalEndpoints(keyTokens.get(6)));
- assertTrue(endPoints.size() == 6);
- assertTrue(endPoints.contains(hosts.get(7)));
- assertTrue(endPoints.contains(hosts.get(8)));
- assertTrue(endPoints.contains(hosts.get(9)));
- assertTrue(endPoints.contains(hosts.get(0)));
- assertTrue(endPoints.contains(hosts.get(1)));
- assertTrue(endPoints.contains(boot2));
-
- // token 75 should to go nodes 8, 9, 0, 1, 2 and boot2
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(7), testStrategy.getNaturalEndpoints(keyTokens.get(7)));
- assertTrue(endPoints.size() == 6);
- assertTrue(endPoints.contains(hosts.get(8)));
- assertTrue(endPoints.contains(hosts.get(9)));
- assertTrue(endPoints.contains(hosts.get(0)));
- assertTrue(endPoints.contains(hosts.get(1)));
- assertTrue(endPoints.contains(hosts.get(2)));
- assertTrue(endPoints.contains(boot2));
-
- // token 85 should go to nodes 9, 0, 1 and 2
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(8), testStrategy.getNaturalEndpoints(keyTokens.get(8)));
- assertTrue(endPoints.size() == 4);
- assertTrue(endPoints.contains(hosts.get(9)));
- assertTrue(endPoints.contains(hosts.get(0)));
- assertTrue(endPoints.contains(hosts.get(1)));
- assertTrue(endPoints.contains(hosts.get(2)));
-
- // token 95 should go to nodes 0, 1 and 2
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(9), testStrategy.getNaturalEndpoints(keyTokens.get(9)));
- assertTrue(endPoints.size() == 3);
- assertTrue(endPoints.contains(hosts.get(0)));
- assertTrue(endPoints.contains(hosts.get(1)));
- assertTrue(endPoints.contains(hosts.get(2)));
-
// Now finish node 6 and node 9 leaving, as well as boot1 (after this node 8 is still
// leaving and boot2 in progress
- ss.onChange(hosts.get(6), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEFT + StorageService.Delimiter + StorageService.LEFT_NORMALLY + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(6))));
- ss.onChange(hosts.get(9), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEFT + StorageService.Delimiter + StorageService.LEFT_NORMALLY + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(9))));
+ ss.onChange(hosts.get(LEAVING[0]), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEFT + StorageService.Delimiter + StorageService.LEFT_NORMALLY + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(LEAVING[0]))));
+ ss.onChange(hosts.get(LEAVING[2]), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEFT + StorageService.Delimiter + StorageService.LEFT_NORMALLY + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(LEAVING[2]))));
ss.onChange(boot1, StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_NORMAL + StorageService.Delimiter + partitioner.getTokenFactory().toString(keyTokens.get(5))));
- // tokens 5, 15 and 25 should go three nodes
- for (int i=0; i<3; ++i)
+ // adjust precalcuated results. this changes what the epected endpoints are.
+ expectedEndpoints.get("Keyspace1").get(new BigIntegerToken("55")).removeAll(makeAddrs("127.0.0.7", "127.0.0.8"));
+ expectedEndpoints.get("Keyspace1").get(new BigIntegerToken("85")).removeAll(makeAddrs("127.0.0.10"));
+ expectedEndpoints.get("Keyspace2").get(new BigIntegerToken("55")).removeAll(makeAddrs("127.0.0.7", "127.0.0.8"));
+ expectedEndpoints.get("Keyspace2").get(new BigIntegerToken("85")).removeAll(makeAddrs("127.0.0.10"));
+ expectedEndpoints.get("Keyspace3").get(new BigIntegerToken("15")).removeAll(makeAddrs("127.0.0.7", "127.0.0.8"));
+ expectedEndpoints.get("Keyspace3").get(new BigIntegerToken("25")).removeAll(makeAddrs("127.0.0.7", "127.0.1.2", "127.0.0.1"));
+ expectedEndpoints.get("Keyspace3").get(new BigIntegerToken("35")).removeAll(makeAddrs("127.0.0.7", "127.0.0.2"));
+ expectedEndpoints.get("Keyspace3").get(new BigIntegerToken("45")).removeAll(makeAddrs("127.0.0.7", "127.0.0.10", "127.0.0.3"));
+ expectedEndpoints.get("Keyspace3").get(new BigIntegerToken("55")).removeAll(makeAddrs("127.0.0.7", "127.0.0.10", "127.0.0.4"));
+ expectedEndpoints.get("Keyspace3").get(new BigIntegerToken("65")).removeAll(makeAddrs("127.0.0.10"));
+ expectedEndpoints.get("Keyspace3").get(new BigIntegerToken("75")).removeAll(makeAddrs("127.0.0.10"));
+ expectedEndpoints.get("Keyspace3").get(new BigIntegerToken("85")).removeAll(makeAddrs("127.0.0.10"));
+ expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("35")).removeAll(makeAddrs("127.0.0.7", "127.0.0.8"));
+ expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("45")).removeAll(makeAddrs("127.0.0.7", "127.0.1.2", "127.0.0.1"));
+ expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("55")).removeAll(makeAddrs("127.0.0.2", "127.0.0.7"));
+ expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("65")).removeAll(makeAddrs("127.0.0.10"));
+ expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("75")).removeAll(makeAddrs("127.0.0.10"));
+ expectedEndpoints.get("Keyspace4").get(new BigIntegerToken("85")).removeAll(makeAddrs("127.0.0.10"));
+
+ for (String table : DatabaseDescriptor.getNonSystemTables())
{
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(i), testStrategy.getNaturalEndpoints(keyTokens.get(i)));
+ for (int i = 0; i < keyTokens.size(); i++)
+ {
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table));
+ assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).size() == endPoints.size());
+ assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).containsAll(endPoints));
+ }
+
+ if (DatabaseDescriptor.getReplicationFactor(table) != 3)
+ continue;
+ // leave this stuff in to guarantee the old tests work the way they were supposed to.
+ // tokens 5, 15 and 25 should go three nodes
+ for (int i=0; i<3; ++i)
+ {
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table));
+ assertTrue(endPoints.size() == 3);
+ assertTrue(endPoints.contains(hosts.get(i+1)));
+ assertTrue(endPoints.contains(hosts.get(i+2)));
+ assertTrue(endPoints.contains(hosts.get(i+3)));
+ }
+
+ // token 35 goes to nodes 4, 5 and boot1
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(3), table, testStrategy.getNaturalEndpoints(keyTokens.get(3), table));
assertTrue(endPoints.size() == 3);
- assertTrue(endPoints.contains(hosts.get(i+1)));
- assertTrue(endPoints.contains(hosts.get(i+2)));
- assertTrue(endPoints.contains(hosts.get(i+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 = testStrategy.getWriteEndpoints(keyTokens.get(4), table, testStrategy.getNaturalEndpoints(keyTokens.get(4), table));
+ 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 = testStrategy.getWriteEndpoints(keyTokens.get(5), table, testStrategy.getNaturalEndpoints(keyTokens.get(5), table));
+ assertTrue(endPoints.size() == 5);
+ assertTrue(endPoints.contains(boot1));
+ assertTrue(endPoints.contains(hosts.get(7)));
+ assertTrue(endPoints.contains(boot2));
+ assertTrue(endPoints.contains(hosts.get(8)));
+ assertTrue(endPoints.contains(hosts.get(0)));
+
+ // token 65 goes to nodes 7, boot2, 8, 0 and 1
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(6), table, testStrategy.getNaturalEndpoints(keyTokens.get(6), table));
+ assertTrue(endPoints.size() == 5);
+ assertTrue(endPoints.contains(hosts.get(7)));
+ assertTrue(endPoints.contains(boot2));
+ assertTrue(endPoints.contains(hosts.get(8)));
+ assertTrue(endPoints.contains(hosts.get(0)));
+ assertTrue(endPoints.contains(hosts.get(1)));
+
+ // token 75 goes to nodes boot2, 8, 0, 1 and 2
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(7), table, testStrategy.getNaturalEndpoints(keyTokens.get(7), table));
+ assertTrue(endPoints.size() == 5);
+ assertTrue(endPoints.contains(boot2));
+ assertTrue(endPoints.contains(hosts.get(8)));
+ assertTrue(endPoints.contains(hosts.get(0)));
+ assertTrue(endPoints.contains(hosts.get(1)));
+ assertTrue(endPoints.contains(hosts.get(2)));
+
+ // token 85 goes to nodes 0, 1 and 2
+ endPoints = testStrategy.getWriteEndpoints(keyTokens.get(8), table, testStrategy.getNaturalEndpoints(keyTokens.get(8), table));
+ 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 = testStrategy.getWriteEndpoints(keyTokens.get(9), table, testStrategy.getNaturalEndpoints(keyTokens.get(9), table));
+ assertTrue(endPoints.size() == 3);
+ assertTrue(endPoints.contains(hosts.get(0)));
+ assertTrue(endPoints.contains(hosts.get(1)));
+ assertTrue(endPoints.contains(hosts.get(2)));
}
- // token 35 goes to nodes 4, 5 and boot1
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(3), testStrategy.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 = testStrategy.getWriteEndpoints(keyTokens.get(4), testStrategy.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 = testStrategy.getWriteEndpoints(keyTokens.get(5), testStrategy.getNaturalEndpoints(keyTokens.get(5)));
- assertTrue(endPoints.size() == 5);
- assertTrue(endPoints.contains(boot1));
- assertTrue(endPoints.contains(hosts.get(7)));
- assertTrue(endPoints.contains(boot2));
- assertTrue(endPoints.contains(hosts.get(8)));
- assertTrue(endPoints.contains(hosts.get(0)));
-
- // token 65 goes to nodes 7, boot2, 8, 0 and 1
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(6), testStrategy.getNaturalEndpoints(keyTokens.get(6)));
- assertTrue(endPoints.size() == 5);
- assertTrue(endPoints.contains(hosts.get(7)));
- assertTrue(endPoints.contains(boot2));
- assertTrue(endPoints.contains(hosts.get(8)));
- assertTrue(endPoints.contains(hosts.get(0)));
- assertTrue(endPoints.contains(hosts.get(1)));
-
- // token 75 goes to nodes boot2, 8, 0, 1 and 2
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(7), testStrategy.getNaturalEndpoints(keyTokens.get(7)));
- assertTrue(endPoints.size() == 5);
- assertTrue(endPoints.contains(boot2));
- assertTrue(endPoints.contains(hosts.get(8)));
- assertTrue(endPoints.contains(hosts.get(0)));
- assertTrue(endPoints.contains(hosts.get(1)));
- assertTrue(endPoints.contains(hosts.get(2)));
-
- // token 85 goes to nodes 0, 1 and 2
- endPoints = testStrategy.getWriteEndpoints(keyTokens.get(8), testStrategy.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 = testStrategy.getWriteEndpoints(keyTokens.get(9), testStrategy.getNaturalEndpoints(keyTokens.get(9)));
- assertTrue(endPoints.size() == 3);
- assertTrue(endPoints.contains(hosts.get(0)));
- assertTrue(endPoints.contains(hosts.get(1)));
- assertTrue(endPoints.contains(hosts.get(2)));
-
ss.setPartitionerUnsafe(oldPartitioner);
ss.setReplicationStrategyUnsafe(oldStrategy);
}
@@ -290,10 +421,10 @@ public class MoveTest
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
- AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, 3);
+ AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, null);
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
- AbstractReplicationStrategy oldStrategy = ss.setReplicationStrategyUnsafe(testStrategy);
+ Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList endPointTokens = new ArrayList();
ArrayList keyTokens = new ArrayList();
@@ -359,10 +490,10 @@ public class MoveTest
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
- AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, 3);
+ AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, null);
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
- AbstractReplicationStrategy oldStrategy = ss.setReplicationStrategyUnsafe(testStrategy);
+ Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList endPointTokens = new ArrayList();
ArrayList keyTokens = new ArrayList();
@@ -403,10 +534,10 @@ public class MoveTest
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
- AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, 3);
+ AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, null);
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
- AbstractReplicationStrategy oldStrategy = ss.setReplicationStrategyUnsafe(testStrategy);
+ Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList endPointTokens = new ArrayList();
ArrayList keyTokens = new ArrayList();
@@ -453,10 +584,10 @@ public class MoveTest
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
- AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, 3);
+ AbstractReplicationStrategy testStrategy = new RackUnawareStrategy(tmd, null);
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
- AbstractReplicationStrategy oldStrategy = ss.setReplicationStrategyUnsafe(testStrategy);
+ Map oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList endPointTokens = new ArrayList();
ArrayList keyTokens = new ArrayList();
@@ -513,4 +644,11 @@ public class MoveTest
assertTrue(ss.getTokenMetadata().isMember(hosts.get(i)));
}
+ private static Collection makeAddrs(String... hosts) throws UnknownHostException
+ {
+ ArrayList addrs = new ArrayList(hosts.length);
+ for (String host : hosts)
+ addrs.add(InetAddress.getByName(host));
+ return addrs;
+ }
}