diff --git a/src/java/org/apache/cassandra/client/RingCache.java b/src/java/org/apache/cassandra/client/RingCache.java index a43fb3755e..137eea656a 100644 --- a/src/java/org/apache/cassandra/client/RingCache.java +++ b/src/java/org/apache/cassandra/client/RingCache.java @@ -24,7 +24,7 @@ import java.util.Set; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Token; -import org.apache.cassandra.locator.IReplicaPlacementStrategy; +import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.locator.TokenMetadata; import org.apache.cassandra.net.EndPoint; import org.apache.cassandra.service.Cassandra; @@ -47,7 +47,7 @@ public class RingCache private Set seeds_ = new HashSet(); final private int port_=DatabaseDescriptor.getThriftPort(); - private volatile IReplicaPlacementStrategy nodePicker_; + private volatile AbstractReplicationStrategy nodePicker_; final private static IPartitioner partitioner_ = DatabaseDescriptor.getPartitioner(); public RingCache() @@ -84,7 +84,7 @@ public class RingCache Class [] parameterTypes = new Class[] { TokenMetadata.class, IPartitioner.class, int.class, int.class}; try { - nodePicker_ = (IReplicaPlacementStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata, partitioner_, DatabaseDescriptor.getReplicationFactor(), port_); + nodePicker_ = (AbstractReplicationStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata, partitioner_, DatabaseDescriptor.getReplicationFactor(), port_); } catch (Exception e) { @@ -102,6 +102,6 @@ public class RingCache public EndPoint[] getEndPoint(String key) { - return nodePicker_.getStorageEndPoints(partitioner_.getToken(key)); + return nodePicker_.getReadStorageEndPoints(partitioner_.getToken(key)); } } diff --git a/src/java/org/apache/cassandra/db/LoadVerbHandler.java b/src/java/org/apache/cassandra/db/LoadVerbHandler.java deleted file mode 100644 index ba212a70db..0000000000 --- a/src/java/org/apache/cassandra/db/LoadVerbHandler.java +++ /dev/null @@ -1,67 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.cassandra.db; - -import org.apache.cassandra.net.EndPoint; -import org.apache.cassandra.net.IVerbHandler; -import org.apache.cassandra.net.Message; -import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.utils.LogUtil; -import org.apache.cassandra.io.DataInputBuffer; -import org.apache.log4j.Logger; - -public class LoadVerbHandler implements IVerbHandler -{ - private static Logger logger_ = Logger.getLogger(LoadVerbHandler.class); - - public void doVerb(Message message) - { - try - { - byte[] body = message.getMessageBody(); - DataInputBuffer buffer = new DataInputBuffer(); - buffer.reset(body, body.length); - RowMutationMessage rmMsg = RowMutationMessage.serializer().deserialize(buffer); - - EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(rmMsg.getRowMutation().key()); - - Message messageInternal = new Message(StorageService.getLocalStorageEndPoint(), - StorageService.mutationStage_, - StorageService.mutationVerbHandler_, - body - ); - - StringBuilder sb = new StringBuilder(); - for(EndPoint endPoint : endpoints) - { - sb.append(endPoint); - MessagingService.getMessagingInstance().sendOneWay(messageInternal, endPoint); - } - if (logger_.isDebugEnabled()) - logger_.debug("Sent data to " + sb.toString()); - } - catch ( Exception e ) - { - if (logger_.isDebugEnabled()) - logger_.debug(LogUtil.throwableToString(e)); - } - } - -} diff --git a/src/java/org/apache/cassandra/db/ReadVerbHandler.java b/src/java/org/apache/cassandra/db/ReadVerbHandler.java index 3d565a52c9..93707059ce 100644 --- a/src/java/org/apache/cassandra/db/ReadVerbHandler.java +++ b/src/java/org/apache/cassandra/db/ReadVerbHandler.java @@ -29,8 +29,7 @@ import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.service.UnavailableException; -import org.apache.cassandra.utils.LogUtil; + import org.apache.log4j.Logger; public class ReadVerbHandler implements IVerbHandler @@ -115,7 +114,7 @@ public class ReadVerbHandler implements IVerbHandler private void doReadRepair(Row row, ReadCommand readCommand) { - List endpoints = StorageService.instance().getNLiveStorageEndPoint(readCommand.key); + List endpoints = StorageService.instance().getLiveReadStorageEndPoints(readCommand.key); /* Remove the local storage endpoint from the list. */ endpoints.remove( StorageService.getLocalStorageEndPoint() ); diff --git a/src/java/org/apache/cassandra/locator/AbstractStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java similarity index 89% rename from src/java/org/apache/cassandra/locator/AbstractStrategy.java rename to src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index 709b17340b..fe85bbcea4 100644 --- a/src/java/org/apache/cassandra/locator/AbstractStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -37,16 +37,16 @@ import org.apache.cassandra.net.EndPoint; * all abstraction that implement the IReplicaPlacementStrategy * interface. */ -public abstract class AbstractStrategy implements IReplicaPlacementStrategy +public abstract class AbstractReplicationStrategy { - protected static final Logger logger_ = Logger.getLogger(AbstractStrategy.class); + protected static final Logger logger_ = Logger.getLogger(AbstractReplicationStrategy.class); protected TokenMetadata tokenMetadata_; protected IPartitioner partitioner_; protected int replicas_; protected int storagePort_; - AbstractStrategy(TokenMetadata tokenMetadata, IPartitioner partitioner, int replicas, int storagePort) + AbstractReplicationStrategy(TokenMetadata tokenMetadata, IPartitioner partitioner, int replicas, int storagePort) { tokenMetadata_ = tokenMetadata; partitioner_ = partitioner; @@ -54,6 +54,21 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy storagePort_ = storagePort; } + public abstract EndPoint[] getWriteStorageEndPoints(Token token); + public abstract EndPoint[] getReadStorageEndPoints(Token token, Map tokenToEndPointMap); + public abstract EndPoint[] getReadStorageEndPoints(Token token); + + /* + * This method returns the hint map. The key is the endpoint + * on which the data is being placed and the value is the + * endpoint which is in the top N. + * Get the map of top N to the live nodes currently. + */ + public Map getHintedStorageEndPoints(Token token) + { + return getHintedMapForEndpoints(getWriteStorageEndPoints(token)); + } + /* * This method changes the ports of the endpoints from * the control port to the storage ports. @@ -94,18 +109,6 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy return endPoint; } - /* - * This method returns the hint map. The key is the endpoint - * on which the data is being placed and the value is the - * endpoint which is in the top N. - * Get the map of top N to the live nodes currently. - */ - public Map getHintedStorageEndPoints(Token token) - { - EndPoint[] topN = getStorageEndPointsForWrite( token ); - return getHintedMapForEndpoints(topN); - } - private Map getHintedMapForEndpoints(EndPoint[] topN) { List liveList = new ArrayList(); @@ -135,6 +138,4 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy } return map; } - - public abstract EndPoint[] getStorageEndPoints(Token token); } diff --git a/src/java/org/apache/cassandra/locator/IReplicaPlacementStrategy.java b/src/java/org/apache/cassandra/locator/IReplicaPlacementStrategy.java deleted file mode 100644 index a53331bf99..0000000000 --- a/src/java/org/apache/cassandra/locator/IReplicaPlacementStrategy.java +++ /dev/null @@ -1,40 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.cassandra.locator; - -import java.util.Map; -import java.math.BigInteger; - -import org.apache.cassandra.dht.Token; -import org.apache.cassandra.net.EndPoint; - - -/* - * This interface has two implementations. One which - * does not respect rack or datacenter awareness and - * the other which does. - */ -public interface IReplicaPlacementStrategy -{ - public EndPoint[] getStorageEndPoints(Token token); - public EndPoint[] getStorageEndPointsForWrite(Token token); - public Map getStorageEndPoints(String[] keys); - public EndPoint[] getStorageEndPoints(Token token, Map tokenToEndPointMap); - public Map getHintedStorageEndPoints(Token token); -} diff --git a/src/java/org/apache/cassandra/locator/RackAwareStrategy.java b/src/java/org/apache/cassandra/locator/RackAwareStrategy.java index 48b5d08b87..24fc0e76f9 100644 --- a/src/java/org/apache/cassandra/locator/RackAwareStrategy.java +++ b/src/java/org/apache/cassandra/locator/RackAwareStrategy.java @@ -21,7 +21,6 @@ package org.apache.cassandra.locator; import java.net.UnknownHostException; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Map; @@ -38,14 +37,14 @@ import org.apache.cassandra.utils.LogUtil; * a node in a different rack in the same datacenter as * the primary. */ -public class RackAwareStrategy extends AbstractStrategy +public class RackAwareStrategy extends AbstractReplicationStrategy { public RackAwareStrategy(TokenMetadata tokenMetadata, IPartitioner partitioner, int replicas, int storagePort) { super(tokenMetadata, partitioner, replicas, storagePort); } - public EndPoint[] getStorageEndPoints(Token token) + public EndPoint[] getReadStorageEndPoints(Token token) { int startIndex; List list = new ArrayList(); @@ -123,24 +122,12 @@ public class RackAwareStrategy extends AbstractStrategy return list.toArray(new EndPoint[list.size()]); } - public Map getStorageEndPoints(String[] keys) - { - Map results = new HashMap(); - - for ( String key : keys ) - { - results.put(key, getStorageEndPoints(partitioner_.getToken(key))); - } - - return results; - } - - public EndPoint[] getStorageEndPoints(Token token, Map tokenToEndPointMap) + public EndPoint[] getReadStorageEndPoints(Token token, Map tokenToEndPointMap) { throw new UnsupportedOperationException("This operation is not currently supported"); } - public EndPoint[] getStorageEndPointsForWrite(Token token) + public EndPoint[] getWriteStorageEndPoints(Token token) { throw new UnsupportedOperationException("Rack-aware bootstrapping not supported"); } diff --git a/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java b/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java index 6ad7ce171b..f482bd3e34 100644 --- a/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java +++ b/src/java/org/apache/cassandra/locator/RackUnawareStrategy.java @@ -20,7 +20,6 @@ package org.apache.cassandra.locator; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Map; @@ -34,19 +33,19 @@ import org.apache.cassandra.net.EndPoint; * returns the 3 nodes that lie right next to each other * on the ring. */ -public class RackUnawareStrategy extends AbstractStrategy +public class RackUnawareStrategy extends AbstractReplicationStrategy { public RackUnawareStrategy(TokenMetadata tokenMetadata, IPartitioner partitioner, int replicas, int storagePort) { super(tokenMetadata, partitioner, replicas, storagePort); } - public EndPoint[] getStorageEndPoints(Token token) + public EndPoint[] getReadStorageEndPoints(Token token) { - return getStorageEndPoints(token, tokenMetadata_.cloneTokenEndPointMap()); + return getReadStorageEndPoints(token, tokenMetadata_.cloneTokenEndPointMap()); } - public EndPoint[] getStorageEndPointsForWrite(Token token) + public EndPoint[] getWriteStorageEndPoints(Token token) { Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); Map bootstrapTokensToEndpointMap = tokenMetadata_.cloneBootstrapNodes(); @@ -64,7 +63,7 @@ public class RackUnawareStrategy extends AbstractStrategy return list.toArray(new EndPoint[list.size()]); } - public EndPoint[] getStorageEndPoints(Token token, Map tokenToEndPointMap) + public EndPoint[] getReadStorageEndPoints(Token token, Map tokenToEndPointMap) { List tokenList = getStorageTokens(token, tokenToEndPointMap, null); List list = new ArrayList(); @@ -115,15 +114,4 @@ public class RackUnawareStrategy extends AbstractStrategy } return tokenList; } - - public Map getStorageEndPoints(String[] keys) - { - Map results = new HashMap(); - - for ( String key : keys ) - { - results.put(key, getStorageEndPoints(partitioner_.getToken(key))); - } - return results; - } } diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 1f0a650298..34b10e67e6 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -113,7 +113,7 @@ public class StorageProxy implements StorageProxyMBean try { // (This is the ZERO consistency level, so user doesn't care if we don't really have N destinations available.) - Map endpointMap = StorageService.instance().getNStorageEndPointMap(rm.key()); + Map endpointMap = StorageService.instance().getHintedStorageEndpointMap(rm.key()); Map messageMap = createWriteMessages(rm, endpointMap); for (Map.Entry entry : messageMap.entrySet()) { @@ -148,7 +148,7 @@ public class StorageProxy implements StorageProxyMBean } try { - Map endpointMap = StorageService.instance().getNStorageEndPointMap(rm.key()); + Map endpointMap = StorageService.instance().getHintedStorageEndpointMap(rm.key()); int blockFor = determineBlockFor(consistency_level); List primaryNodes = getUnhintedNodes(endpointMap); if (primaryNodes.size() < blockFor) // guarantee blockFor = W live nodes. @@ -294,7 +294,7 @@ public class StorageProxy implements StorageProxyMBean for (ReadCommand command: commands) { - EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(command.key); + EndPoint[] endpoints = StorageService.instance().getReadStorageEndPoints(command.key); boolean foundLocal = Arrays.asList(endpoints).contains(StorageService.getLocalStorageEndPoint()); //TODO: Throw InvalidRequest if we're in bootstrap mode? if (foundLocal && !StorageService.instance().isBootstrapMode()) @@ -358,7 +358,7 @@ public class StorageProxy implements StorageProxyMBean DatabaseDescriptor.getQuorum(), readResponseResolver); EndPoint dataPoint = StorageService.instance().findSuitableEndPoint(command.key); - List endpointList = new ArrayList(Arrays.asList(StorageService.instance().getNStorageEndPoint(command.key))); + List endpointList = new ArrayList(Arrays.asList(StorageService.instance().getReadStorageEndPoints(command.key))); /* Remove the local storage endpoint from the list. */ endpointList.remove(dataPoint); EndPoint[] endPoints = new EndPoint[endpointList.size() + 1]; @@ -466,7 +466,7 @@ public class StorageProxy implements StorageProxyMBean { /* This is the primary */ EndPoint dataPoint = StorageService.instance().findSuitableEndPoint(key); - List replicas = new ArrayList( StorageService.instance().getNLiveStorageEndPoint(key) ); + List replicas = new ArrayList( StorageService.instance().getLiveReadStorageEndPoints(key) ); replicas.remove(dataPoint); /* Get the messages to be sent index 0 is the data messages and index 1 is the digest message */ Message[] message = messages.get(key); @@ -506,7 +506,7 @@ public class StorageProxy implements StorageProxyMBean List rows = new ArrayList(); for (ReadCommand command: commands) { - List endpoints = StorageService.instance().getNLiveStorageEndPoint(command.key); + List endpoints = StorageService.instance().getLiveReadStorageEndPoints(command.key); /* Remove the local storage endpoint from the list. */ endpoints.remove(StorageService.getLocalStorageEndPoint()); // TODO: throw a thrift exception if we do not have N nodes diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 0e2dd245d5..a16f0e4773 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -66,7 +66,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto /* All verb handler identifiers */ public final static String mutationVerbHandler_ = "ROW-MUTATION-VERB-HANDLER"; public final static String tokenVerbHandler_ = "TOKEN-VERB-HANDLER"; - public final static String loadVerbHandler_ = "LOAD-VERB-HANDLER"; public final static String binaryVerbHandler_ = "BINARY-VERB-HANDLER"; public final static String readRepairVerbHandler_ = "READ-REPAIR-VERB-HANDLER"; public final static String readVerbHandler_ = "ROW-READ-VERB-HANDLER"; @@ -78,12 +77,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public final static String bsMetadataVerbHandler_ = "BS-METADATA-VERB-HANDLER"; public final static String rangeVerbHandler_ = "RANGE-VERB-HANDLER"; - public static enum ConsistencyLevel - { - WEAK, - STRONG - } - private static StorageService instance_; /* Used to lock the factory for creation of StorageService instance */ private static Lock createLock_ = new ReentrantLock(); @@ -104,7 +97,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public static IPartitioner getPartitioner() { return partitioner_; } - + static { partitioner_ = DatabaseDescriptor.getPartitioner(); @@ -183,7 +176,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto /* This is the entity that tracks load information of all nodes in the cluster */ private StorageLoadBalancer storageLoadBalancer_; /* We use this interface to determine where replicas need to be placed */ - private IReplicaPlacementStrategy nodePicker_; + private AbstractReplicationStrategy nodePicker_; /* Are we starting this node in bootstrap mode? */ private boolean isBootstrapMode; private Set bootstrapSet; @@ -241,7 +234,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto /* register the verb handlers */ MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.tokenVerbHandler_, new TokenUpdateVerbHandler()); MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.binaryVerbHandler_, new BinaryVerbHandler()); - MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.loadVerbHandler_, new LoadVerbHandler()); MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.mutationVerbHandler_, new RowMutationVerbHandler()); MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.readRepairVerbHandler_, new ReadRepairVerbHandler()); MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.readVerbHandler_, new ReadVerbHandler()); @@ -268,7 +260,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto Class [] parameterTypes = new Class[] { TokenMetadata.class, IPartitioner.class, int.class, int.class}; try { - nodePicker_ = (IReplicaPlacementStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort()); + nodePicker_ = (AbstractReplicationStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort()); } catch (Exception e) { @@ -377,7 +369,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto Map> rangeToEndPointMap = new HashMap>(); for ( Range range : ranges ) { - EndPoint[] endpoints = getNStorageEndPoint(range.right()); + EndPoint[] endpoints = nodePicker_.getReadStorageEndPoints(range.right()); rangeToEndPointMap.put(range, new ArrayList( Arrays.asList(endpoints) ) ); } if (logger_.isDebugEnabled()) @@ -399,7 +391,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto Map> rangeToEndPointMap = new HashMap>(); for ( Range range : ranges ) { - EndPoint[] endpoints = getNStorageEndPoint(range.right(), tokenToEndPointMap); + EndPoint[] endpoints = nodePicker_.getReadStorageEndPoints(range.right(), tokenToEndPointMap); rangeToEndPointMap.put(range, new ArrayList( Arrays.asList(endpoints) ) ); } if (logger_.isDebugEnabled()) @@ -417,8 +409,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto { Map> endPointToRangesMap = new HashMap>(); Map tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap(); - Collection mbrs = tokenToEndPointMap.values(); - for ( EndPoint mbr : mbrs ) + for (EndPoint mbr : tokenToEndPointMap.values()) { endPointToRangesMap.put(mbr, getRangesForEndPoint(mbr)); } @@ -952,16 +943,10 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto * @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[] getReadStorageEndPoints(String key) { - return nodePicker_.getStorageEndPoints(partitioner_.getToken(key)); - } - - private Map getNStorageEndPoints(String[] keys) - { - return nodePicker_.getStorageEndPoints(keys); - } - + return nodePicker_.getReadStorageEndPoints(partitioner_.getToken(key)); + } /** * This method attempts to return N endpoints that are responsible for storing the @@ -970,10 +955,10 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto * @param key - key for which we need to find the endpoint return value - * the endpoint responsible for this key */ - public List getNLiveStorageEndPoint(String key) + public List getLiveReadStorageEndPoints(String key) { List liveEps = new ArrayList(); - EndPoint[] endpoints = getNStorageEndPoint(key); + EndPoint[] endpoints = getReadStorageEndPoints(key); for ( EndPoint endpoint : endpoints ) { @@ -991,42 +976,18 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto * @param key - key for which we need to find the endpoint return value - * the endpoint responsible for this key */ - public Map getNStorageEndPointMap(String key) + public Map getHintedStorageEndpointMap(String key) { return nodePicker_.getHintedStorageEndPoints(partitioner_.getToken(key)); } - /** - * This method returns the N endpoints that are responsible for storing the - * specified token i.e for replication. - * - * @param token - position on the ring - */ - public EndPoint[] getNStorageEndPoint(Token token) - { - return nodePicker_.getStorageEndPoints(token); - } - - /** - * This method returns the N endpoints that are responsible for storing the - * specified token i.e for replication and are based on the token to endpoint - * mapping that is passed in. - * - * @param token - position on the ring - * @param tokens - w/o the following tokens in the token list - */ - protected EndPoint[] getNStorageEndPoint(Token token, Map tokenToEndPointMap) - { - return nodePicker_.getStorageEndPoints(token, tokenToEndPointMap); - } - /** * This function finds the most suitable endpoint given a key. * It checks for locality and alive test. */ public EndPoint findSuitableEndPoint(String key) throws IOException { - EndPoint[] endpoints = getNStorageEndPoint(key); + EndPoint[] endpoints = getReadStorageEndPoints(key); for(EndPoint endPoint: endpoints) { if(endPoint.equals(StorageService.getLocalStorageEndPoint())) @@ -1057,68 +1018,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } return null; } - - /* - * TODO: - * This is used by the incomplete multiget implementation. Need to rewrite - * this to use findSuitableEndPoint above instead of copy/paste - */ - public Map findSuitableEndPoints(String[] keys) throws IOException - { - Map suitableEndPoints = new HashMap(); - Map results = getNStorageEndPoints(keys); - for ( String key : keys ) - { - EndPoint[] endpoints = results.get(key); - /* indicates if we have to move on to the next key */ - boolean moveOn = false; - for(EndPoint endPoint: endpoints) - { - if(endPoint.equals(StorageService.getLocalStorageEndPoint())) - { - suitableEndPoints.put(key, endPoint); - moveOn = true; - break; - } - } - - if ( moveOn ) - continue; - - int j = 0; - for ( ; j < endpoints.length; ++j ) - { - if ( StorageService.instance().isInSameDataCenter(endpoints[j]) && FailureDetector.instance().isAlive(endpoints[j]) ) - { - if (logger_.isDebugEnabled()) - logger_.debug("EndPoint " + endpoints[j] + " is in the same data center as local storage endpoint."); - suitableEndPoints.put(key, endpoints[j]); - moveOn = true; - break; - } - } - - if ( moveOn ) - continue; - - // We have tried to be really nice but looks like there are no servers - // in the local data center that are alive and can service this request so - // just send it to the first alive guy and see if we get anything. - j = 0; - for ( ; j < endpoints.length; ++j ) - { - if ( FailureDetector.instance().isAlive(endpoints[j]) ) - { - if (logger_.isDebugEnabled()) - logger_.debug("EndPoint " + endpoints[j] + " is alive so get data from it."); - suitableEndPoints.put(key, endpoints[j]); - break; - } - } - } - return suitableEndPoints; - } - + Map getLiveEndPointMap() { return tokenMetadata_.cloneTokenEndPointMap(); diff --git a/src/java/org/apache/cassandra/tools/KeyChecker.java b/src/java/org/apache/cassandra/tools/KeyChecker.java index d844561f74..ad3ec76225 100644 --- a/src/java/org/apache/cassandra/tools/KeyChecker.java +++ b/src/java/org/apache/cassandra/tools/KeyChecker.java @@ -40,7 +40,7 @@ public class KeyChecker */ private static boolean checkIfProcessKey(String key) { - EndPoint[] endPoints = StorageService.instance().getNStorageEndPoint(key); + EndPoint[] endPoints = StorageService.instance().getReadStorageEndPoints(key); EndPoint localEndPoint = StorageService.getLocalStorageEndPoint(); for(EndPoint endPoint : endPoints) { diff --git a/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java b/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java index 28043c492f..8e8e9f712c 100644 --- a/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java +++ b/test/unit/org/apache/cassandra/locator/RackUnawareStrategyTest.java @@ -40,7 +40,7 @@ public class RackUnawareStrategyTest { TokenMetadata tmd = new TokenMetadata(); IPartitioner partitioner = new RandomPartitioner(); - IReplicaPlacementStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000); + AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000); List endPointTokens = new ArrayList(); List keyTokens = new ArrayList(); @@ -56,7 +56,7 @@ public class RackUnawareStrategyTest { TokenMetadata tmd = new TokenMetadata(); IPartitioner partitioner = new OrderPreservingPartitioner(); - IReplicaPlacementStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000); + AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000); List endPointTokens = new ArrayList(); List keyTokens = new ArrayList(); @@ -69,7 +69,7 @@ public class RackUnawareStrategyTest // 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 testGetStorageEndPoints(TokenMetadata tmd, IReplicaPlacementStrategy strategy, Token[] endPointTokens, Token[] keyTokens) + private void testGetStorageEndPoints(TokenMetadata tmd, AbstractReplicationStrategy strategy, Token[] endPointTokens, Token[] keyTokens) { List hosts = new ArrayList(); for (int i = 0; i < endPointTokens.length; i++) @@ -81,7 +81,7 @@ public class RackUnawareStrategyTest for (int i = 0; i < keyTokens.length; i++) { - EndPoint[] endPoints = strategy.getStorageEndPoints(keyTokens[i]); + EndPoint[] endPoints = strategy.getReadStorageEndPoints(keyTokens[i]); assertEquals(3, endPoints.length); for (int j = 0; j < endPoints.length; j++) { @@ -95,7 +95,7 @@ public class RackUnawareStrategyTest { TokenMetadata tmd = new TokenMetadata(); IPartitioner partitioner = new RandomPartitioner(); - IReplicaPlacementStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000); + AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000); Token[] endPointTokens = new Token[5]; Token[] keyTokens = new Token[5]; @@ -121,7 +121,7 @@ public class RackUnawareStrategyTest for (int i = 0; i < keyTokens.length; i++) { - EndPoint[] endPoints = strategy.getStorageEndPointsForWrite(keyTokens[i]); + EndPoint[] endPoints = strategy.getWriteStorageEndPoints(keyTokens[i]); assertTrue(endPoints.length >=3); List endPointsList = Arrays.asList(endPoints);