You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2009/07/31 20:29:53 UTC
svn commit: r799684 - in /incubator/cassandra/trunk:
src/java/org/apache/cassandra/db/ src/java/org/apache/cassandra/locator/
src/java/org/apache/cassandra/service/ src/java/org/apache/cassandra/tools/
test/unit/org/apache/cassandra/db/ test/unit/org/a...
Author: jbellis
Date: Fri Jul 31 18:29:52 2009
New Revision: 799684
URL: http://svn.apache.org/viewvc?rev=799684&view=rev
Log:
Make range queries continue on the next node when necessary.
Patch by Arin Sarkissian; reviewed by jbellis for CASSANDRA-212
Modified:
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/LoadVerbHandler.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/RangeReply.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractStrategy.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/IReplicaPlacementStrategy.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/RangeVerbHandler.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/tools/KeyChecker.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/CompactionsTest.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/OneCompactionTest.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/LoadVerbHandler.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/LoadVerbHandler.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/LoadVerbHandler.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/LoadVerbHandler.java Fri Jul 31 18:29:52 2009
@@ -44,7 +44,7 @@
buffer.reset(body, body.length);
RowMutationMessage rmMsg = RowMutationMessage.serializer().deserialize(buffer);
- EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(rmMsg.getRowMutation().key());
+ EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(rmMsg.getRowMutation().key(), 0);
Message messageInternal = new Message(StorageService.getLocalStorageEndPoint(),
StorageService.mutationStage_,
@@ -68,4 +68,4 @@
}
}
-}
+}
\ No newline at end of file
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/RangeReply.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/RangeReply.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/RangeReply.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/RangeReply.java Fri Jul 31 18:29:52 2009
@@ -29,25 +29,22 @@
public class RangeReply
{
public final List<String> keys;
+ public final boolean rangeCompletedLocally;
- public RangeReply(List<String> keys)
+ public RangeReply(List<String> keys, boolean rangeCompletedLocally)
{
this.keys = Collections.unmodifiableList(keys);
+ this.rangeCompletedLocally = rangeCompletedLocally;
}
- public Message getReply(Message originalMessage)
+ public Message getReply(Message originalMessage) throws IOException
{
DataOutputBuffer dob = new DataOutputBuffer();
+ dob.writeBoolean(rangeCompletedLocally);
+
for (String key : keys)
{
- try
- {
- dob.writeUTF(key);
- }
- catch (IOException e)
- {
- throw new RuntimeException(e);
- }
+ dob.writeUTF(key);
}
byte[] data = Arrays.copyOf(dob.getData(), dob.getLength());
return originalMessage.getReply(StorageService.getLocalStorageEndPoint(), data);
@@ -56,7 +53,9 @@
public static RangeReply read(byte[] body) throws IOException
{
DataInputBuffer bufIn = new DataInputBuffer();
+ boolean rangeCompletedLocally;
bufIn.reset(body, body.length);
+ rangeCompletedLocally = bufIn.readBoolean();
List<String> keys = new ArrayList<String>();
while (bufIn.getPosition() < body.length)
@@ -64,6 +63,6 @@
keys.add(bufIn.readUTF());
}
- return new RangeReply(keys);
+ return new RangeReply(keys, rangeCompletedLocally);
}
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java Fri Jul 31 18:29:52 2009
@@ -656,7 +656,7 @@
* @param maxResults
* @return list of keys between startWith and stopAt
*/
- public List<String> getKeyRange(String columnFamily, final String startWith, final String stopAt, int maxResults)
+ public RangeReply getKeyRange(String columnFamily, final String startWith, final String stopAt, int maxResults)
throws IOException, ExecutionException, InterruptedException
{
assert getColumnFamilyStore(columnFamily) != null : columnFamily;
@@ -672,7 +672,7 @@
}
}
- private List<String> getKeyRangeUnsafe(final String cfName, final String startWith, final String stopAt, int maxResults) throws IOException, ExecutionException, InterruptedException
+ private RangeReply getKeyRangeUnsafe(final String cfName, final String startWith, final String stopAt, int maxResults) throws IOException, ExecutionException, InterruptedException
{
// (OPP key decoration is a no-op so using the "decorated" comparator against raw keys is fine)
final Comparator<String> comparator = StorageService.getPartitioner().getDecoratedKeyComparator();
@@ -729,10 +729,12 @@
// pull keys out of the CollatedIterator. checking tombstone status is expensive,
// so we set an arbitrary limit on how many we'll do at once.
List<String> keys = new ArrayList<String>();
+ boolean rangeCompletedLocally = false;
for (String current : reduced)
{
if (!stopAt.isEmpty() && comparator.compare(stopAt, current) < 0)
{
+ rangeCompletedLocally = true;
break;
}
// make sure there is actually non-tombstone content associated w/ this key
@@ -744,10 +746,11 @@
}
if (keys.size() >= maxResults)
{
+ rangeCompletedLocally = true;
break;
}
}
- return keys;
+ return new RangeReply(keys, rangeCompletedLocally);
}
finally
{
@@ -765,4 +768,4 @@
{
return dataDirPath + File.separator + tableName + File.separator + SNAPSHOT_SUBDIR_NAME + File.separator + snapshotName;
}
-}
+}
\ No newline at end of file
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractStrategy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractStrategy.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractStrategy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/AbstractStrategy.java Fri Jul 31 18:29:52 2009
@@ -104,7 +104,7 @@
{
List<EndPoint> liveList = new ArrayList<EndPoint>();
Map<EndPoint, EndPoint> map = new HashMap<EndPoint, EndPoint>();
- EndPoint[] topN = getStorageEndPoints( token );
+ EndPoint[] topN = getStorageEndPoints( token, 0);
for( int i = 0 ; i < topN.length ; i++)
{
@@ -131,5 +131,5 @@
return map;
}
- public abstract EndPoint[] getStorageEndPoints(Token token);
-}
+ public abstract EndPoint[] getStorageEndPoints(Token token, int offset);
+}
\ No newline at end of file
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/IReplicaPlacementStrategy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/IReplicaPlacementStrategy.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/IReplicaPlacementStrategy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/IReplicaPlacementStrategy.java Fri Jul 31 18:29:52 2009
@@ -33,8 +33,8 @@
*/
public interface IReplicaPlacementStrategy
{
- public EndPoint[] getStorageEndPoints(Token token);
- public Map<String, EndPoint[]> getStorageEndPoints(String[] keys);
- public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap);
+ public EndPoint[] getStorageEndPoints(Token token, int offset);
+ public Map<String, EndPoint[]> getStorageEndPoints(String[] keys, int offset);
+ public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap, int offset);
public Map<EndPoint, EndPoint> getHintedStorageEndPoints(Token token);
-}
+}
\ No newline at end of file
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackAwareStrategy.java Fri Jul 31 18:29:52 2009
@@ -45,7 +45,7 @@
super(tokenMetadata, partitioner, replicas, storagePort);
}
- public EndPoint[] getStorageEndPoints(Token token)
+ public EndPoint[] getStorageEndPoints(Token token, int offset)
{
int startIndex;
List<EndPoint> list = new ArrayList<EndPoint>();
@@ -123,19 +123,19 @@
return list.toArray(new EndPoint[list.size()]);
}
- public Map<String, EndPoint[]> getStorageEndPoints(String[] keys)
+ public Map<String, EndPoint[]> getStorageEndPoints(String[] keys, int offset)
{
Map<String, EndPoint[]> results = new HashMap<String, EndPoint[]>();
for ( String key : keys )
{
- results.put(key, getStorageEndPoints(partitioner_.getInitialToken(key)));
+ results.put(key, getStorageEndPoints(partitioner_.getInitialToken(key), offset));
}
return results;
}
- public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap)
+ public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap, int offset)
{
throw new UnsupportedOperationException("This operation is not currently supported");
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java Fri Jul 31 18:29:52 2009
@@ -41,12 +41,12 @@
super(tokenMetadata, partitioner, replicas, storagePort);
}
- public EndPoint[] getStorageEndPoints(Token token)
+ public EndPoint[] getStorageEndPoints(Token token, int offset)
{
- return getStorageEndPoints(token, tokenMetadata_.cloneTokenEndPointMap());
+ return getStorageEndPoints(token, tokenMetadata_.cloneTokenEndPointMap(), offset);
}
- public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap)
+ public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap, int offset)
{
int startIndex;
List<EndPoint> list = new ArrayList<EndPoint>();
@@ -61,8 +61,9 @@
index = 0;
}
int totalNodes = tokens.size();
+ int realIndex = (index + offset) % totalNodes;
// Add the node at the index by default
- list.add(tokenToEndPointMap.get(tokens.get(index)));
+ list.add(tokenToEndPointMap.get(tokens.get(realIndex)));
foundCount++;
startIndex = (index + 1)%totalNodes;
// If we found N number of nodes we are good. This loop will just exit. Otherwise just
@@ -79,15 +80,15 @@
return list.toArray(new EndPoint[list.size()]);
}
- public Map<String, EndPoint[]> getStorageEndPoints(String[] keys)
+ public Map<String, EndPoint[]> getStorageEndPoints(String[] keys, int offset)
{
Map<String, EndPoint[]> results = new HashMap<String, EndPoint[]>();
for ( String key : keys )
{
- results.put(key, getStorageEndPoints(partitioner_.getInitialToken(key)));
+ results.put(key, getStorageEndPoints(partitioner_.getInitialToken(key), offset));
}
return results;
}
-}
+}
\ No newline at end of file
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/RangeVerbHandler.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/RangeVerbHandler.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/RangeVerbHandler.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/RangeVerbHandler.java Fri Jul 31 18:29:52 2009
@@ -18,8 +18,6 @@
*/
package org.apache.cassandra.service;
-import java.util.List;
-
import org.apache.cassandra.db.RangeCommand;
import org.apache.cassandra.db.RangeReply;
import org.apache.cassandra.db.Table;
@@ -31,19 +29,19 @@
{
public void doVerb(Message message)
{
- List<String> keys;
+ RangeReply rangeReply;
try
{
RangeCommand command = RangeCommand.read(message);
Table table = Table.open(command.table);
- keys = table.getKeyRange(command.columnFamily, command.startWith, command.stopAt, command.maxResults);
+
+ rangeReply = table.getKeyRange(command.columnFamily, command.startWith, command.stopAt, command.maxResults);
+ Message response = rangeReply.getReply(message);
+ MessagingService.getMessagingInstance().sendOneWay(response, message.getFrom());
}
catch (Exception e)
{
throw new RuntimeException(e);
}
-
- Message response = new RangeReply(keys).getReply(message);
- MessagingService.getMessagingInstance().sendOneWay(response, message.getFrom());
}
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java Fri Jul 31 18:29:52 2009
@@ -164,7 +164,7 @@
}
try
{
- EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(rm.key());
+ EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(rm.key(), 0);
if (endpoints.length < (DatabaseDescriptor.getReplicationFactor() / 2) + 1)
{
throw new UnavailableException();
@@ -312,7 +312,7 @@
long startTime = System.currentTimeMillis();
Row row;
- EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(command.key);
+ EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(command.key, 0);
if (consistency_level == ConsistencyLevel.ONE)
{
@@ -413,7 +413,7 @@
DatabaseDescriptor.getQuorum(),
readResponseResolver);
EndPoint dataPoint = StorageService.instance().findSuitableEndPoint(command.key);
- List<EndPoint> endpointList = new ArrayList<EndPoint>(Arrays.asList(StorageService.instance().getNStorageEndPoint(command.key)));
+ List<EndPoint> endpointList = new ArrayList<EndPoint>(Arrays.asList(StorageService.instance().getNStorageEndPoint(command.key, 0)));
/* Remove the local storage endpoint from the list. */
endpointList.remove(dataPoint);
EndPoint[] endPoints = new EndPoint[endpointList.size() + 1];
@@ -634,15 +634,52 @@
static List<String> getKeyRange(RangeCommand command)
{
long startTime = System.currentTimeMillis();
+ int endpointOffset = 0;
+ List<String> allKeys = new ArrayList<String>();
+ int maxResults = command.maxResults;
+
try
{
- EndPoint endPoint = StorageService.instance().findSuitableEndPoint(command.startWith);
- IAsyncResult iar = MessagingService.getMessagingInstance().sendRR(command.getMessage(), endPoint);
+ EndPoint endPoint = StorageService.instance().findSuitableEndPoint(command.startWith, endpointOffset);
+ String firstEndpoint = endPoint.toString();
+
+ do
+ {
+ IAsyncResult iar = MessagingService.getMessagingInstance().sendRR(command.getMessage(), endPoint);
+
+ // read response
+ byte[] responseBody = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
+ RangeReply rangeReply = RangeReply.read(responseBody);
+ List<String> rangeKeys = rangeReply.keys;
+
+ // deal with key overlaps
+ if (allKeys.size() > 0 && rangeKeys != null && rangeKeys.size() > 0 && allKeys.get(allKeys.size() - 1).equals(rangeKeys.get(0)))
+ {
+ allKeys.remove(allKeys.size() - 1);
+ allKeys.addAll(rangeKeys);
+ }
+ else if (rangeKeys.size() > 0)
+ {
+ allKeys.addAll(rangeKeys);
+ }
+
+ if (allKeys.size() >= maxResults || rangeReply.rangeCompletedLocally)
+ {
+ break;
+ }
+
+ String newStartAt = (allKeys.size() > 0) ? allKeys.get(allKeys.size() - 1) : command.stopAt;
- // read response
- // TODO send more requests if we need to span multiple nodes
- byte[] responseBody = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
- return RangeReply.read(responseBody).keys;
+ command = new RangeCommand(command.table, command.columnFamily,
+ newStartAt, command.stopAt,
+ command.maxResults - rangeKeys.size());
+
+ endPoint = StorageService.instance().findSuitableEndPoint(command.startWith, ++endpointOffset);
+ } while (!endPoint.toString().equals(firstEndpoint));
+
+ return (allKeys.size() > maxResults)
+ ? allKeys.subList(0, maxResults)
+ : allKeys;
}
catch (Exception e)
{
@@ -683,4 +720,4 @@
{
return writeStats.size();
}
-}
+}
\ No newline at end of file
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java Fri Jul 31 18:29:52 2009
@@ -931,16 +931,20 @@
* @param key - key for which we need to find the endpoint return value -
* the endpoint responsible for this key
*/
- public EndPoint[] getNStorageEndPoint(String key)
+ public EndPoint[] getNStorageEndPoint(String key, int offset)
{
- return nodePicker_.getStorageEndPoints(partitioner_.getInitialToken(key));
+ return nodePicker_.getStorageEndPoints(partitioner_.getInitialToken(key), offset);
}
- private Map<String, EndPoint[]> getNStorageEndPoints(String[] keys)
+ private Map<String, EndPoint[]> getNStorageEndPoints(String[] keys, int offset)
{
- return nodePicker_.getStorageEndPoints(keys);
+ return nodePicker_.getStorageEndPoints(keys, offset);
}
+ private Map<String, EndPoint[]> getNStorageEndPoints(String[] keys)
+ {
+ return getNStorageEndPoints(keys, 0);
+ }
/**
* This method attempts to return N endpoints that are responsible for storing the
@@ -952,7 +956,7 @@
public List<EndPoint> getNLiveStorageEndPoint(String key)
{
List<EndPoint> liveEps = new ArrayList<EndPoint>();
- EndPoint[] endpoints = getNStorageEndPoint(key);
+ EndPoint[] endpoints = getNStorageEndPoint(key, 0);
for ( EndPoint endpoint : endpoints )
{
@@ -983,7 +987,7 @@
*/
public EndPoint[] getNStorageEndPoint(Token token)
{
- return nodePicker_.getStorageEndPoints(token);
+ return nodePicker_.getStorageEndPoints(token, 0);
}
/**
@@ -996,16 +1000,21 @@
*/
protected EndPoint[] getNStorageEndPoint(Token token, Map<Token, EndPoint> tokenToEndPointMap)
{
- return nodePicker_.getStorageEndPoints(token, tokenToEndPointMap);
+ return nodePicker_.getStorageEndPoints(token, tokenToEndPointMap, 0);
+ }
+
+ public EndPoint findSuitableEndPoint(String key) throws IOException
+ {
+ return findSuitableEndPoint(key, 0);
}
/**
* This function finds the most suitable endpoint given a key.
* It checks for locality and alive test.
*/
- public EndPoint findSuitableEndPoint(String key) throws IOException
+ public EndPoint findSuitableEndPoint(String key, int offset) throws IOException
{
- EndPoint[] endpoints = getNStorageEndPoint(key);
+ EndPoint[] endpoints = getNStorageEndPoint(key, offset);
for(EndPoint endPoint: endpoints)
{
if(endPoint.equals(StorageService.getLocalStorageEndPoint()))
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/tools/KeyChecker.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/tools/KeyChecker.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/tools/KeyChecker.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/tools/KeyChecker.java Fri Jul 31 18:29:52 2009
@@ -40,7 +40,7 @@
*/
private static boolean checkIfProcessKey(String key)
{
- EndPoint[] endPoints = StorageService.instance().getNStorageEndPoint(key);
+ EndPoint[] endPoints = StorageService.instance().getNStorageEndPoint(key, 0);
EndPoint localEndPoint = StorageService.getLocalStorageEndPoint();
for(EndPoint endPoint : endPoints)
{
@@ -96,4 +96,4 @@
System.out.println("DONE checking keys ...");
raf.close();
}
-}
+}
\ No newline at end of file
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/CompactionsTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/CompactionsTest.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/CompactionsTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/CompactionsTest.java Fri Jul 31 18:29:52 2009
@@ -52,7 +52,7 @@
inserted.add(key);
}
store.forceBlockingFlush();
- assertEquals(table.getKeyRange("Standard1", "", "", 10000).size(), inserted.size());
+ assertEquals(table.getKeyRange("Standard1", "", "", 10000).keys.size(), inserted.size());
}
while (true)
{
@@ -64,6 +64,6 @@
{
store.doCompaction(store.getSSTables().size());
}
- assertEquals(table.getKeyRange("Standard1", "", "", 10000).size(), inserted.size());
+ assertEquals(table.getKeyRange("Standard1", "", "", 10000).keys.size(), inserted.size());
}
}
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/OneCompactionTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/OneCompactionTest.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/OneCompactionTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/OneCompactionTest.java Fri Jul 31 18:29:52 2009
@@ -45,12 +45,12 @@
rm.apply();
inserted.add(key);
store.forceBlockingFlush();
- assertEquals(table.getKeyRange(columnFamilyName, "", "", 10000).size(), inserted.size());
+ assertEquals(table.getKeyRange(columnFamilyName, "", "", 10000).keys.size(), inserted.size());
}
Future<Integer> ft = MinorCompactionManager.instance().submit(store, 2);
ft.get();
assertEquals(1, store.getSSTables().size());
- assertEquals(table.getKeyRange(columnFamilyName, "", "", 10000).size(), inserted.size());
+ assertEquals(table.getKeyRange(columnFamilyName, "", "", 10000).keys.size(), inserted.size());
}
@Test
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManager2Test.java Fri Jul 31 18:29:52 2009
@@ -34,7 +34,7 @@
table1.getColumnFamilyStore("Standard1").clearUnsafe();
RecoveryManager.doRecovery();
- Set<String> foundKeys = new HashSet<String>(table1.getKeyRange("Standard1", "", "", 1000));
+ Set<String> foundKeys = new HashSet<String>(table1.getKeyRange("Standard1", "", "", 1000).keys);
assert keys.equals(foundKeys);
}
}
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java?rev=799684&r1=799683&r2=799684&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java Fri Jul 31 18:29:52 2009
@@ -78,7 +78,7 @@
for (int i = 0; i < keyTokens.length; i++)
{
- EndPoint[] endPoints = strategy.getStorageEndPoints(keyTokens[i]);
+ EndPoint[] endPoints = strategy.getStorageEndPoints(keyTokens[i], 0);
assert endPoints.length == 3;
for (int j = 0; j < endPoints.length; j++)
{