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

rename EndPoint -> Endpoint, part 1. patch by Erick Tryzelaar; reviewed by jbellis for CASSANDRA-994 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@935594 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-04-19 14:36:24 +00:00
parent 997c06ffe8
commit 18b2a16f0a
43 changed files with 586 additions and 586 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
conf/cassandra.xml
View File

@ -46,13 +46,13 @@
<AutoBootstrap>false</AutoBootstrap>
<!--
~ EndPointSnitch: Setting this to the class that implements
~ EndpointSnitch: Setting this to the class that implements
~ AbstractEndpointSnitch, which lets Cassandra know enough
~ about your network topology to route requests efficiently.
~ Out of the box, Cassandra provides org.apache.cassandra.locator.EndPointSnitch,
~ and PropertyFileEndPointSnitch is available in contrib/.
-->
<EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
<EndpointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndpointSnitch>
<!--
~ Keyspaces and ColumnFamilies:

2
contrib/circuit/src/org/apache/cassandra/contrib/circuit/RingModel.java
View File

@ -101,7 +101,7 @@ public class RingModel
"Invalid ObjectName? Please report this as a bug.", e);
}
Map<Range, List<String>> rangeMap = ssProxy.getRangeToEndPointMap(null);
Map<Range, List<String>> rangeMap = ssProxy.getRangeToEndpointMap(null);
List<Range> ranges = new ArrayList<Range>(rangeMap.keySet());
Collections.sort(ranges);

8
contrib/pig/storage-conf.xml
View File

@ -127,13 +127,13 @@
<ReplicationFactor>1</ReplicationFactor>
<!--
~ EndPointSnitch: Setting this to the class that implements
~ EndpointSnitch: Setting this to the class that implements
~ AbstractEndpointSnitch, which lets Cassandra know enough
~ about your network topology to route requests efficiently.
~ Out of the box, Cassandra provides org.apache.cassandra.locator.EndPointSnitch,
~ and PropertyFileEndPointSnitch is available in contrib/.
~ Out of the box, Cassandra provides org.apache.cassandra.locator.EndpointSnitch,
~ and PropertyFileEndpointSnitch is available in contrib/.
-->
<EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
<EndpointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndpointSnitch>
</Keyspace>
</Keyspaces>

22
contrib/property_snitch/src/java/org/apache/cassandra/locator/PropertyFileEndPointSnitch.java
View File

@ -80,12 +80,12 @@ public class PropertyFileEndPointSnitch extends EndPointSnitch implements Proper
/**
* Get the raw information about an end point
*
* @param endPoint endPoint to process
* @param endpoint endpoint to process
*
* @return a array of string with the first index being the data center and the second being the rack
*/
public String[] getEndPointInfo(InetAddress endPoint) {
String key = endPoint.toString();
public String[] getEndpointInfo(InetAddress endpoint) {
String key = endpoint.toString();
String value = hostProperties.getProperty(key);
if (value == null)
{
@ -104,22 +104,22 @@ public class PropertyFileEndPointSnitch extends EndPointSnitch implements Proper
/**
* Return the data center for which an endpoint resides in
*
* @param endPoint the endPoint to process
* @param endpoint the endpoint to process
* @return string of data center
*/
public String getDataCenterForEndPoint(InetAddress endPoint) {
return getEndPointInfo(endPoint)[0];
public String getDataCenterForEndpoint(InetAddress endpoint) {
return getEndpointInfo(endpoint)[0];
}
/**
* Return the rack for which an endpoint resides in
*
* @param endPoint the endPoint to process
* @param endpoint the endpoint to process
*
* @return string of rack
*/
public String getRackForEndPoint(InetAddress endPoint) {
return getEndPointInfo(endPoint)[1];
public String getRackForEndpoint(InetAddress endpoint) {
return getEndpointInfo(endpoint)[1];
}
@Override
@ -129,7 +129,7 @@ public class PropertyFileEndPointSnitch extends EndPointSnitch implements Proper
{
return super.isInSameDataCenter(host, host2);
}
return getDataCenterForEndPoint(host).equals(getDataCenterForEndPoint(host2));
return getDataCenterForEndpoint(host).equals(getDataCenterForEndpoint(host2));
}
@Override
@ -143,7 +143,7 @@ public class PropertyFileEndPointSnitch extends EndPointSnitch implements Proper
{
return false;
}
return getRackForEndPoint(host).equals(getRackForEndPoint(host2));
return getRackForEndpoint(host).equals(getRackForEndpoint(host2));
}
public String displayConfiguration() {

4
contrib/word_count/storage-conf.xml
View File

@ -127,13 +127,13 @@
<ReplicationFactor>1</ReplicationFactor>
<!--
~ EndPointSnitch: Setting this to the class that implements
~ EndpointSnitch: Setting this to the class that implements
~ AbstractEndpointSnitch, which lets Cassandra know enough
~ about your network topology to route requests efficiently.
~ Out of the box, Cassandra provides org.apache.cassandra.locator.EndPointSnitch,
~ and PropertyFileEndPointSnitch is available in contrib/.
-->
<EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
<EndpointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndpointSnitch>
</Keyspace>
</Keyspaces>

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

@ -65,10 +65,10 @@ public class RingCache
this.keyspace = keyspace;
DatabaseDescriptor.loadSchemas();
refreshEndPointMap();
refreshEndpointMap();
}
public void refreshEndPointMap()
public void refreshEndpointMap()
{
for (String seed : seeds_)
{
@ -109,7 +109,7 @@ public class RingCache
}
}
public List<InetAddress> getEndPoint(byte[] key)
public List<InetAddress> getEndpoint(byte[] key)
{
if (tokenMetadata == null)
throw new RuntimeException("Must refresh endpoints before looking up a key.");

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

@ -377,12 +377,12 @@ public class DatabaseDescriptor
}
/* end point snitch */
String endPointSnitchClassName = xmlUtils.getNodeValue("/Storage/EndPointSnitch");
if (endPointSnitchClassName == null)
String endpointSnitchClassName = xmlUtils.getNodeValue("/Storage/EndpointSnitch");
if (endpointSnitchClassName == null)
{
throw new ConfigurationException("Missing endpointsnitch directive");
}
snitch = createEndpointSnitch(endPointSnitchClassName);
snitch = createEndpointSnitch(endpointSnitchClassName);
/* snapshot-before-compaction. defaults to false */
String sbc = xmlUtils.getNodeValue("/Storage/SnapshotBeforeCompaction");
@ -524,17 +524,17 @@ public class DatabaseDescriptor
}
}
private static IEndPointSnitch createEndpointSnitch(String endPointSnitchClassName) throws ConfigurationException
private static IEndPointSnitch createEndpointSnitch(String endpointSnitchClassName) throws ConfigurationException
{
IEndPointSnitch snitch;
Class cls;
try
{
cls = Class.forName(endPointSnitchClassName);
cls = Class.forName(endpointSnitchClassName);
}
catch (ClassNotFoundException e)
{
throw new ConfigurationException("Unable to load endpointsnitch class " + endPointSnitchClassName);
throw new ConfigurationException("Unable to load endpointsnitch class " + endpointSnitchClassName);
}
Constructor ctor;
try
@ -543,7 +543,7 @@ public class DatabaseDescriptor
}
catch (NoSuchMethodException e)
{
throw new ConfigurationException("No default constructor found in " + endPointSnitchClassName);
throw new ConfigurationException("No default constructor found in " + endpointSnitchClassName);
}
try
{
@ -551,17 +551,17 @@ public class DatabaseDescriptor
}
catch (InstantiationException e)
{
throw new ConfigurationException("endpointsnitch class " + endPointSnitchClassName + "is abstract");
throw new ConfigurationException("endpointsnitch class " + endpointSnitchClassName + "is abstract");
}
catch (IllegalAccessException e)
{
throw new ConfigurationException("Access to " + endPointSnitchClassName + " constructor was rejected");
throw new ConfigurationException("Access to " + endpointSnitchClassName + " constructor was rejected");
}
catch (InvocationTargetException e)
{
if (e.getCause() instanceof ConfigurationException)
throw (ConfigurationException)e.getCause();
throw new ConfigurationException("Error instantiating " + endPointSnitchClassName + " " + e.getMessage());
throw new ConfigurationException("Error instantiating " + endpointSnitchClassName + " " + e.getMessage());
}
return snitch;
}
@ -872,7 +872,7 @@ public class DatabaseDescriptor
return partitioner;
}
public static IEndPointSnitch getEndPointSnitch()
public static IEndPointSnitch getEndpointSnitch()
{
return snitch;
}

28
src/java/org/apache/cassandra/db/HintedHandOffManager.java
View File

@ -104,14 +104,14 @@ public class HintedHandOffManager
}, "Hint delivery").start();
}
private static boolean sendMessage(InetAddress endPoint, String tableName, byte[] key) throws IOException
private static boolean sendMessage(InetAddress endpoint, String tableName, byte[] key) throws IOException
{
if (!Gossiper.instance.isKnownEndpoint(endPoint))
if (!Gossiper.instance.isKnownEndpoint(endpoint))
{
logger_.warn("Hints found for endpoint " + endPoint + " which is not part of the gossip network. discarding.");
logger_.warn("Hints found for endpoint " + endpoint + " which is not part of the gossip network. discarding.");
return true;
}
if (!FailureDetector.instance.isAlive(endPoint))
if (!FailureDetector.instance.isAlive(endpoint))
{
return false;
}
@ -127,7 +127,7 @@ public class HintedHandOffManager
}
Message message = rm.makeRowMutationMessage();
WriteResponseHandler responseHandler = new WriteResponseHandler(1, tableName);
MessagingService.instance.sendRR(message, new InetAddress[] { endPoint }, responseHandler);
MessagingService.instance.sendRR(message, new InetAddress[] { endpoint }, responseHandler);
try
{
@ -140,7 +140,7 @@ public class HintedHandOffManager
return true;
}
private static void deleteEndPoint(byte[] endpointAddress, String tableName, byte[] key, long timestamp) throws IOException
private static void deleteEndpoint(byte[] endpointAddress, String tableName, byte[] key, long timestamp) throws IOException
{
RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, tableName.getBytes(UTF8));
rm.delete(new QueryPath(HINTS_CF, key, endpointAddress), timestamp);
@ -189,7 +189,7 @@ public class HintedHandOffManager
{
if (sendMessage(InetAddress.getByAddress(endpoint.name()), tableName, keyBytes))
{
deleteEndPoint(endpoint.name(), tableName, keyColumn.name(), System.currentTimeMillis());
deleteEndpoint(endpoint.name(), tableName, keyColumn.name(), System.currentTimeMillis());
deleted++;
}
}
@ -223,12 +223,12 @@ public class HintedHandOffManager
|| (hintColumnFamily.getSortedColumns().size() == 1 && hintColumnFamily.getColumn(startColumn) != null);
}
private static void deliverHintsToEndpoint(InetAddress endPoint) throws IOException, DigestMismatchException, InvalidRequestException, TimeoutException
private static void deliverHintsToEndpoint(InetAddress endpoint) throws IOException, DigestMismatchException, InvalidRequestException, TimeoutException
{
if (logger_.isDebugEnabled())
logger_.debug("Started hinted handoff for endPoint " + endPoint);
logger_.debug("Started hinted handoff for endpoint " + endpoint);
byte[] targetEPBytes = endPoint.getAddress();
byte[] targetEPBytes = endpoint.getAddress();
// 1. Scan through all the keys that we need to handoff
// 2. For each key read the list of recipients if the endpoint matches send
// 3. Delete that recipient from the key if write was successful
@ -249,14 +249,14 @@ public class HintedHandOffManager
{
byte[] keyBytes = keyColumn.name();
Collection<IColumn> endpoints = keyColumn.getSubColumns();
for (IColumn hintEndPoint : endpoints)
for (IColumn hintEndpoint : endpoints)
{
if (Arrays.equals(hintEndPoint.name(), targetEPBytes) && sendMessage(endPoint, tableName, keyBytes))
if (Arrays.equals(hintEndpoint.name(), targetEPBytes) && sendMessage(endpoint, tableName, keyBytes))
{
if (endpoints.size() == 1)
deleteHintKey(tableName, keyColumn.name());
else
deleteEndPoint(hintEndPoint.name(), tableName, keyColumn.name(), System.currentTimeMillis());
deleteEndpoint(hintEndpoint.name(), tableName, keyColumn.name(), System.currentTimeMillis());
break;
}
}
@ -267,7 +267,7 @@ public class HintedHandOffManager
}
if (logger_.isDebugEnabled())
logger_.debug("Finished hinted handoff for endpoint " + endPoint);
logger_.debug("Finished hinted handoff for endpoint " + endpoint);
}
/** called when a keyspace is dropped or rename. newTable==null in the case of a drop. */

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

@ -156,7 +156,7 @@ public class BootStrapper
{
if (range.contains(myRange))
{
List<InetAddress> preferred = DatabaseDescriptor.getEndPointSnitch().getSortedListByProximity(address, rangeAddresses.get(range));
List<InetAddress> preferred = DatabaseDescriptor.getEndpointSnitch().getSortedListByProximity(address, rangeAddresses.get(range));
myRangeAddresses.putAll(myRange, preferred);
break;
}

2
src/java/org/apache/cassandra/gms/FailureDetector.java
View File

@ -120,7 +120,7 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean
return true;
/* Incoming port is assumed to be the Storage port. We need to change it to the control port */
EndPointState epState = Gossiper.instance.getEndPointStateForEndPoint(ep);
EndPointState epState = Gossiper.instance.getEndpointStateForEndpoint(ep);
return epState.isAlive();
}

18
src/java/org/apache/cassandra/gms/GossipDigest.java
View File

@ -40,7 +40,7 @@ public class GossipDigest implements Comparable<GossipDigest>
serializer_ = new GossipDigestSerializer();
}
InetAddress endPoint_;
InetAddress endpoint_;
int generation_;
int maxVersion_;
@ -49,16 +49,16 @@ public class GossipDigest implements Comparable<GossipDigest>
return serializer_;
}
GossipDigest(InetAddress endPoint, int generation, int maxVersion)
GossipDigest(InetAddress endpoint, int generation, int maxVersion)
{
endPoint_ = endPoint;
endpoint_ = endpoint;
generation_ = generation;
maxVersion_ = maxVersion;
}
InetAddress getEndPoint()
InetAddress getEndpoint()
{
return endPoint_;
return endpoint_;
}
int getGeneration()
@ -81,7 +81,7 @@ public class GossipDigest implements Comparable<GossipDigest>
public String toString()
{
StringBuilder sb = new StringBuilder();
sb.append(endPoint_);
sb.append(endpoint_);
sb.append(":");
sb.append(generation_);
sb.append(":");
@ -94,16 +94,16 @@ class GossipDigestSerializer implements ICompactSerializer<GossipDigest>
{
public void serialize(GossipDigest gDigest, DataOutputStream dos) throws IOException
{
CompactEndPointSerializationHelper.serialize(gDigest.endPoint_, dos);
CompactEndPointSerializationHelper.serialize(gDigest.endpoint_, dos);
dos.writeInt(gDigest.generation_);
dos.writeInt(gDigest.maxVersion_);
}
public GossipDigest deserialize(DataInputStream dis) throws IOException
{
InetAddress endPoint = CompactEndPointSerializationHelper.deserialize(dis);
InetAddress endpoint = CompactEndPointSerializationHelper.deserialize(dis);
int generation = dis.readInt();
int version = dis.readInt();
return new GossipDigest(endPoint, generation, version);
return new GossipDigest(endpoint, generation, version);
}
}

2
src/java/org/apache/cassandra/gms/GossipDigestAck2Message.java
View File

@ -52,7 +52,7 @@ class GossipDigestAck2Message
epStateMap_ = epStateMap;
}
Map<InetAddress, EndPointState> getEndPointStateMap()
Map<InetAddress, EndPointState> getEndpointStateMap()
{
return epStateMap_;
}

2
src/java/org/apache/cassandra/gms/GossipDigestAckMessage.java
View File

@ -60,7 +60,7 @@ class GossipDigestAckMessage
return gDigestList_;
}
Map<InetAddress, EndPointState> getEndPointStateMap()
Map<InetAddress, EndPointState> getEndpointStateMap()
{
return epStateMap_;
}

180
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -55,7 +55,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
synchronized( Gossiper.instance )
{
/* Update the local heartbeat counter. */
endPointStateMap_.get(localEndPoint_).getHeartBeatState().updateHeartBeat();
endpointStateMap_.get(localEndpoint_).getHeartBeatState().updateHeartBeat();
List<GossipDigest> gDigests = new ArrayList<GossipDigest>();
Gossiper.instance.makeRandomGossipDigest(gDigests);
@ -106,7 +106,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
public static final Gossiper instance = new Gossiper();
private Timer gossipTimer_;
private InetAddress localEndPoint_;
private InetAddress localEndpoint_;
private long aVeryLongTime_;
private long FatClientTimeout_;
private Random random_ = new Random();
@ -124,13 +124,13 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
private Set<InetAddress> seeds_ = new HashSet<InetAddress>();
/* map where key is the endpoint and value is the state associated with the endpoint */
Map<InetAddress, EndPointState> endPointStateMap_ = new Hashtable<InetAddress, EndPointState>();
Map<InetAddress, EndPointState> endpointStateMap_ = new Hashtable<InetAddress, EndPointState>();
/* map where key is endpoint and value is timestamp when this endpoint was removed from
* gossip. We will ignore any gossip regarding these endpoints for Streaming.RING_DELAY time
* after removal to prevent nodes from falsely reincarnating during the time when removal
* gossip gets propagated to all nodes */
Map<InetAddress, Long> justRemovedEndPoints_ = new Hashtable<InetAddress, Long>();
Map<InetAddress, Long> justRemovedEndpoints_ = new Hashtable<InetAddress, Long>();
private Gossiper()
{
@ -157,7 +157,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
public Set<InetAddress> getLiveMembers()
{
Set<InetAddress> liveMbrs = new HashSet<InetAddress>(liveEndpoints_);
liveMbrs.add(localEndPoint_);
liveMbrs.add(localEndpoint_);
return liveMbrs;
}
@ -174,7 +174,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
*/
public void convict(InetAddress endpoint)
{
EndPointState epState = endPointStateMap_.get(endpoint);
EndPointState epState = endpointStateMap_.get(endpoint);
if (epState.isAlive())
{
logger_.info("InetAddress {} is now dead.", endpoint);
@ -182,7 +182,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
}
}
int getMaxEndPointStateVersion(EndPointState epState)
int getMaxEndpointStateVersion(EndPointState epState)
{
List<Integer> versions = new ArrayList<Integer>();
versions.add( epState.getHeartBeatState().getHeartBeatVersion() );
@ -214,13 +214,13 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
/**
* Removes the endpoint completely from Gossip
*/
public void removeEndPoint(InetAddress endpoint)
public void removeEndpoint(InetAddress endpoint)
{
liveEndpoints_.remove(endpoint);
unreachableEndpoints_.remove(endpoint);
endPointStateMap_.remove(endpoint);
endpointStateMap_.remove(endpoint);
FailureDetector.instance.remove(endpoint);
justRemovedEndPoints_.put(endpoint, System.currentTimeMillis());
justRemovedEndpoints_.put(endpoint, System.currentTimeMillis());
}
/**
@ -233,25 +233,25 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
void makeRandomGossipDigest(List<GossipDigest> gDigests)
{
/* Add the local endpoint state */
EndPointState epState = endPointStateMap_.get(localEndPoint_);
EndPointState epState = endpointStateMap_.get(localEndpoint_);
int generation = epState.getHeartBeatState().getGeneration();
int maxVersion = getMaxEndPointStateVersion(epState);
gDigests.add( new GossipDigest(localEndPoint_, generation, maxVersion) );
int maxVersion = getMaxEndpointStateVersion(epState);
gDigests.add( new GossipDigest(localEndpoint_, generation, maxVersion) );
List<InetAddress> endpoints = new ArrayList<InetAddress>(endPointStateMap_.keySet());
List<InetAddress> endpoints = new ArrayList<InetAddress>(endpointStateMap_.keySet());
Collections.shuffle(endpoints, random_);
for (InetAddress endPoint : endpoints)
for (InetAddress endpoint : endpoints)
{
epState = endPointStateMap_.get(endPoint);
epState = endpointStateMap_.get(endpoint);
if (epState != null)
{
generation = epState.getHeartBeatState().getGeneration();
maxVersion = getMaxEndPointStateVersion(epState);
gDigests.add(new GossipDigest(endPoint, generation, maxVersion));
maxVersion = getMaxEndpointStateVersion(epState);
gDigests.add(new GossipDigest(endpoint, generation, maxVersion));
}
else
{
gDigests.add(new GossipDigest(endPoint, 0, 0));
gDigests.add(new GossipDigest(endpoint, 0, 0));
}
}
@ -268,12 +268,12 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
public boolean isKnownEndpoint(InetAddress endpoint)
{
return endPointStateMap_.containsKey(endpoint);
return endpointStateMap_.containsKey(endpoint);
}
public int getCurrentGenerationNumber(InetAddress endpoint)
{
return endPointStateMap_.get(endpoint).getHeartBeatState().getGeneration();
return endpointStateMap_.get(endpoint).getHeartBeatState().getGeneration();
}
Message makeGossipDigestSynMessage(List<GossipDigest> gDigests) throws IOException
@ -282,7 +282,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
ByteArrayOutputStream bos = new ByteArrayOutputStream(Gossiper.MAX_GOSSIP_PACKET_SIZE);
DataOutputStream dos = new DataOutputStream( bos );
GossipDigestSynMessage.serializer().serialize(gDigestMessage, dos);
return new Message(localEndPoint_, StageManager.GOSSIP_STAGE, StorageService.Verb.GOSSIP_DIGEST_SYN, bos.toByteArray());
return new Message(localEndpoint_, StageManager.GOSSIP_STAGE, StorageService.Verb.GOSSIP_DIGEST_SYN, bos.toByteArray());
}
Message makeGossipDigestAckMessage(GossipDigestAckMessage gDigestAckMessage) throws IOException
@ -292,7 +292,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
GossipDigestAckMessage.serializer().serialize(gDigestAckMessage, dos);
if (logger_.isTraceEnabled())
logger_.trace("@@@@ Size of GossipDigestAckMessage is " + bos.toByteArray().length);
return new Message(localEndPoint_, StageManager.GOSSIP_STAGE, StorageService.Verb.GOSSIP_DIGEST_ACK, bos.toByteArray());
return new Message(localEndpoint_, StageManager.GOSSIP_STAGE, StorageService.Verb.GOSSIP_DIGEST_ACK, bos.toByteArray());
}
Message makeGossipDigestAck2Message(GossipDigestAck2Message gDigestAck2Message) throws IOException
@ -300,7 +300,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
ByteArrayOutputStream bos = new ByteArrayOutputStream(Gossiper.MAX_GOSSIP_PACKET_SIZE);
DataOutputStream dos = new DataOutputStream(bos);
GossipDigestAck2Message.serializer().serialize(gDigestAck2Message, dos);
return new Message(localEndPoint_, StageManager.GOSSIP_STAGE, StorageService.Verb.GOSSIP_DIGEST_ACK2, bos.toByteArray());
return new Message(localEndpoint_, StageManager.GOSSIP_STAGE, StorageService.Verb.GOSSIP_DIGEST_ACK2, bos.toByteArray());
}
/**
@ -314,9 +314,9 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
{
int size = epSet.size();
/* Generate a random number from 0 -> size */
List<InetAddress> liveEndPoints = new ArrayList<InetAddress>(epSet);
List<InetAddress> liveEndpoints = new ArrayList<InetAddress>(epSet);
int index = (size == 1) ? 0 : random_.nextInt(size);
InetAddress to = liveEndPoints.get(index);
InetAddress to = liveEndpoints.get(index);
if (logger_.isTraceEnabled())
logger_.trace("Sending a GossipDigestSynMessage to {} ...", to);
MessagingService.instance.sendOneWay(message, to);
@ -337,12 +337,12 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
/* Sends a Gossip message to an unreachable member */
void doGossipToUnreachableMember(Message message)
{
double liveEndPoints = liveEndpoints_.size();
double unreachableEndPoints = unreachableEndpoints_.size();
if ( unreachableEndPoints > 0 )
double liveEndpoints = liveEndpoints_.size();
double unreachableEndpoints = unreachableEndpoints_.size();
if ( unreachableEndpoints > 0 )
{
/* based on some probability */
double prob = unreachableEndPoints / (liveEndPoints + 1);
double prob = unreachableEndpoints / (liveEndpoints + 1);
double randDbl = random_.nextDouble();
if ( randDbl < prob )
sendGossip(message, unreachableEndpoints_);
@ -355,7 +355,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
int size = seeds_.size();
if ( size > 0 )
{
if ( size == 1 && seeds_.contains(localEndPoint_) )
if ( size == 1 && seeds_.contains(localEndpoint_) )
{
return;
}
@ -379,14 +379,14 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
{
long now = System.currentTimeMillis();
Set<InetAddress> eps = endPointStateMap_.keySet();
Set<InetAddress> eps = endpointStateMap_.keySet();
for ( InetAddress endpoint : eps )
{
if ( endpoint.equals(localEndPoint_) )
if ( endpoint.equals(localEndpoint_) )
continue;
FailureDetector.instance.interpret(endpoint);
EndPointState epState = endPointStateMap_.get(endpoint);
EndPointState epState = endpointStateMap_.get(endpoint);
if ( epState != null )
{
long duration = now - epState.getUpdateTimestamp();
@ -400,7 +400,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
else
{
logger_.info("FatClient " + endpoint + " has been silent for " + FatClientTimeout_ + "ms, removing from gossip");
removeEndPoint(endpoint);
removeEndpoint(endpoint);
}
}
@ -410,33 +410,33 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
}
}
if (!justRemovedEndPoints_.isEmpty())
if (!justRemovedEndpoints_.isEmpty())
{
Hashtable<InetAddress, Long> copy = new Hashtable<InetAddress, Long>(justRemovedEndPoints_);
Hashtable<InetAddress, Long> copy = new Hashtable<InetAddress, Long>(justRemovedEndpoints_);
for (Map.Entry<InetAddress, Long> entry : copy.entrySet())
{
if ((now - entry.getValue()) > StorageService.RING_DELAY)
{
if (logger_.isDebugEnabled())
logger_.debug(StorageService.RING_DELAY + " elapsed, " + entry.getKey() + " gossip quarantine over");
justRemovedEndPoints_.remove(entry.getKey());
justRemovedEndpoints_.remove(entry.getKey());
}
}
}
}
}
EndPointState getEndPointStateForEndPoint(InetAddress ep)
EndPointState getEndpointStateForEndpoint(InetAddress ep)
{
return endPointStateMap_.get(ep);
return endpointStateMap_.get(ep);
}
synchronized EndPointState getStateForVersionBiggerThan(InetAddress forEndpoint, int version)
{
if (logger_.isTraceEnabled())
logger_.trace("Scanning for state greater than " + version + " for " + forEndpoint);
EndPointState epState = endPointStateMap_.get(forEndpoint);
EndPointState reqdEndPointState = null;
EndPointState epState = endpointStateMap_.get(forEndpoint);
EndPointState reqdEndpointState = null;
if ( epState != null )
{
@ -451,7 +451,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
int localHbVersion = epState.getHeartBeatState().getHeartBeatVersion();
if ( localHbVersion > version )
{
reqdEndPointState = new EndPointState(epState.getHeartBeatState());
reqdEndpointState = new EndPointState(epState.getHeartBeatState());
}
Map<String, ApplicationState> appStateMap = epState.getApplicationStateMap();
/* Accumulate all application states whose versions are greater than "version" variable */
@ -460,18 +460,18 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
ApplicationState appState = entry.getValue();
if ( appState.getStateVersion() > version )
{
if ( reqdEndPointState == null )
if ( reqdEndpointState == null )
{
reqdEndPointState = new EndPointState(epState.getHeartBeatState());
reqdEndpointState = new EndPointState(epState.getHeartBeatState());
}
final String key = entry.getKey();
if (logger_.isTraceEnabled())
logger_.trace("Adding state " + key + ": " + appState.getValue());
reqdEndPointState.addApplicationState(key, appState);
reqdEndpointState.addApplicationState(key, appState);
}
}
}
return reqdEndPointState;
return reqdEndpointState;
}
void notifyFailureDetector(List<GossipDigest> gDigests)
@ -479,29 +479,29 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
IFailureDetector fd = FailureDetector.instance;
for ( GossipDigest gDigest : gDigests )
{
EndPointState localEndPointState = endPointStateMap_.get(gDigest.endPoint_);
EndPointState localEndpointState = endpointStateMap_.get(gDigest.endpoint_);
/*
* If the local endpoint state exists then report to the FD only
* if the versions workout.
*/
if ( localEndPointState != null )
if ( localEndpointState != null )
{
int localGeneration = endPointStateMap_.get(gDigest.endPoint_).getHeartBeatState().generation_;
int localGeneration = endpointStateMap_.get(gDigest.endpoint_).getHeartBeatState().generation_;
int remoteGeneration = gDigest.generation_;
if ( remoteGeneration > localGeneration )
{
fd.report(gDigest.endPoint_);
fd.report(gDigest.endpoint_);
continue;
}
if ( remoteGeneration == localGeneration )
{
int localVersion = getMaxEndPointStateVersion(localEndPointState);
//int localVersion = endPointStateMap_.get(gDigest.endPoint_).getHeartBeatState().getHeartBeatVersion();
int localVersion = getMaxEndpointStateVersion(localEndpointState);
//int localVersion = endpointStateMap_.get(gDigest.endpoint_).getHeartBeatState().getHeartBeatVersion();
int remoteVersion = gDigest.maxVersion_;
if ( remoteVersion > localVersion )
{
fd.report(gDigest.endPoint_);
fd.report(gDigest.endpoint_);
}
}
}
@ -514,16 +514,16 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
for (Entry<InetAddress, EndPointState> entry : remoteEpStateMap.entrySet())
{
InetAddress endpoint = entry.getKey();
EndPointState remoteEndPointState = entry.getValue();
EndPointState localEndPointState = endPointStateMap_.get(endpoint);
EndPointState remoteEndpointState = entry.getValue();
EndPointState localEndpointState = endpointStateMap_.get(endpoint);
/*
* If the local endpoint state exists then report to the FD only
* if the versions workout.
*/
if ( localEndPointState != null )
if ( localEndpointState != null )
{
int localGeneration = localEndPointState.getHeartBeatState().generation_;
int remoteGeneration = remoteEndPointState.getHeartBeatState().generation_;
int localGeneration = localEndpointState.getHeartBeatState().generation_;
int remoteGeneration = remoteEndpointState.getHeartBeatState().generation_;
if ( remoteGeneration > localGeneration )
{
fd.report(endpoint);
@ -532,9 +532,9 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
if ( remoteGeneration == localGeneration )
{
int localVersion = getMaxEndPointStateVersion(localEndPointState);
//int localVersion = localEndPointState.getHeartBeatState().getHeartBeatVersion();
int remoteVersion = remoteEndPointState.getHeartBeatState().getHeartBeatVersion();
int localVersion = getMaxEndpointStateVersion(localEndpointState);
//int localVersion = localEndpointState.getHeartBeatState().getHeartBeatVersion();
int remoteVersion = remoteEndpointState.getHeartBeatState().getHeartBeatVersion();
if ( remoteVersion > localVersion )
{
fd.report(endpoint);
@ -557,7 +557,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
private void handleNewJoin(InetAddress ep, EndPointState epState)
{
if (justRemovedEndPoints_.containsKey(ep))
if (justRemovedEndpoints_.containsKey(ep))
return;
logger_.info("Node {} is now part of the cluster", ep);
handleMajorStateChange(ep, epState, false);
@ -578,11 +578,11 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
*
* @param ep endpoint
* @param epState EndPointState for the endpoint
* @param isKnownNode is this node familiar to us already (present in endPointStateMap)
* @param isKnownNode is this node familiar to us already (present in endpointStateMap)
*/
private void handleMajorStateChange(InetAddress ep, EndPointState epState, boolean isKnownNode)
{
endPointStateMap_.put(ep, epState);
endpointStateMap_.put(ep, epState);
isAlive(ep, epState, isKnownNode);
for (IEndPointStateChangeSubscriber subscriber : subscribers_)
subscriber.onJoin(ep, epState);
@ -593,10 +593,10 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
for (Entry<InetAddress, EndPointState> entry : epStateMap.entrySet())
{
InetAddress ep = entry.getKey();
if ( ep.equals( localEndPoint_ ) )
if ( ep.equals( localEndpoint_ ) )
continue;
EndPointState localEpStatePtr = endPointStateMap_.get(ep);
EndPointState localEpStatePtr = endpointStateMap_.get(ep);
EndPointState remoteState = entry.getValue();
/*
If state does not exist just add it. If it does then add it only if the version
@ -614,8 +614,8 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
else if ( remoteGeneration == localGeneration )
{
/* manage the membership state */
int localMaxVersion = getMaxEndPointStateVersion(localEpStatePtr);
int remoteMaxVersion = getMaxEndPointStateVersion(remoteState);
int localMaxVersion = getMaxEndpointStateVersion(localEpStatePtr);
int remoteMaxVersion = getMaxEndpointStateVersion(remoteState);
if ( remoteMaxVersion > localMaxVersion )
{
markAlive(ep, localEpStatePtr);
@ -732,15 +732,15 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
void requestAll(GossipDigest gDigest, List<GossipDigest> deltaGossipDigestList, int remoteGeneration)
{
/* We are here since we have no data for this endpoint locally so request everthing. */
deltaGossipDigestList.add( new GossipDigest(gDigest.getEndPoint(), remoteGeneration, 0) );
deltaGossipDigestList.add( new GossipDigest(gDigest.getEndpoint(), remoteGeneration, 0) );
}
/* Send all the data with version greater than maxRemoteVersion */
void sendAll(GossipDigest gDigest, Map<InetAddress, EndPointState> deltaEpStateMap, int maxRemoteVersion)
{
EndPointState localEpStatePtr = getStateForVersionBiggerThan(gDigest.getEndPoint(), maxRemoteVersion) ;
EndPointState localEpStatePtr = getStateForVersionBiggerThan(gDigest.getEndpoint(), maxRemoteVersion) ;
if ( localEpStatePtr != null )
deltaEpStateMap.put(gDigest.getEndPoint(), localEpStatePtr);
deltaEpStateMap.put(gDigest.getEndpoint(), localEpStatePtr);
}
/*
@ -754,7 +754,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
int remoteGeneration = gDigest.getGeneration();
int maxRemoteVersion = gDigest.getMaxVersion();
/* Get state associated with the end point in digest */
EndPointState epStatePtr = endPointStateMap_.get(gDigest.getEndPoint());
EndPointState epStatePtr = endpointStateMap_.get(gDigest.getEndpoint());
/*
Here we need to fire a GossipDigestAckMessage. If we have some data associated with this endpoint locally
then we follow the "if" path of the logic. If we have absolutely nothing for this endpoint we need to
@ -764,7 +764,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
{
int localGeneration = epStatePtr.getHeartBeatState().getGeneration();
/* get the max version of all keys in the state associated with this endpoint */
int maxLocalVersion = getMaxEndPointStateVersion(epStatePtr);
int maxLocalVersion = getMaxEndpointStateVersion(epStatePtr);
if ( remoteGeneration == localGeneration && maxRemoteVersion == maxLocalVersion )
continue;
@ -789,7 +789,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
*/
if ( maxRemoteVersion > maxLocalVersion )
{
deltaGossipDigestList.add( new GossipDigest(gDigest.getEndPoint(), remoteGeneration, maxLocalVersion) );
deltaGossipDigestList.add( new GossipDigest(gDigest.getEndpoint(), remoteGeneration, maxLocalVersion) );
}
if ( maxRemoteVersion < maxLocalVersion )
{
@ -810,27 +810,27 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
* Start the gossiper with the generation # retrieved from the System
* table
*/
public void start(InetAddress localEndPoint, int generationNbr)
public void start(InetAddress localEndpoint, int generationNbr)
{
localEndPoint_ = localEndPoint;
localEndpoint_ = localEndpoint;
/* Get the seeds from the config and initialize them. */
Set<InetAddress> seedHosts = DatabaseDescriptor.getSeeds();
for (InetAddress seed : seedHosts)
{
if (seed.equals(localEndPoint))
if (seed.equals(localEndpoint))
continue;
seeds_.add(seed);
}
/* initialize the heartbeat state for this localEndPoint */
EndPointState localState = endPointStateMap_.get(localEndPoint_);
/* initialize the heartbeat state for this localEndpoint */
EndPointState localState = endpointStateMap_.get(localEndpoint_);
if ( localState == null )
{
HeartBeatState hbState = new HeartBeatState(generationNbr);
localState = new EndPointState(hbState);
localState.isAlive(true);
localState.isAGossiper(true);
endPointStateMap_.put(localEndPoint_, localState);
endpointStateMap_.put(localEndpoint_, localState);
}
/* starts a timer thread */
@ -840,7 +840,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
public synchronized void addLocalApplicationState(String key, ApplicationState appState)
{
assert !StorageService.instance.isClientMode();
EndPointState epState = endPointStateMap_.get(localEndPoint_);
EndPointState epState = endpointStateMap_.get(localEndpoint_);
assert epState != null;
epState.addApplicationState(key, appState);
}
@ -909,7 +909,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
Map<InetAddress, GossipDigest> epToDigestMap = new HashMap<InetAddress, GossipDigest>();
for ( GossipDigest gDigest : gDigestList )
{
epToDigestMap.put(gDigest.getEndPoint(), gDigest);
epToDigestMap.put(gDigest.getEndpoint(), gDigest);
}
/*
@ -919,9 +919,9 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
List<GossipDigest> diffDigests = new ArrayList<GossipDigest>();
for ( GossipDigest gDigest : gDigestList )
{
InetAddress ep = gDigest.getEndPoint();
EndPointState epState = Gossiper.instance.getEndPointStateForEndPoint(ep);
int version = (epState != null) ? Gossiper.instance.getMaxEndPointStateVersion( epState ) : 0;
InetAddress ep = gDigest.getEndpoint();
EndPointState epState = Gossiper.instance.getEndpointStateForEndpoint(ep);
int version = (epState != null) ? Gossiper.instance.getMaxEndpointStateVersion( epState ) : 0;
int diffVersion = Math.abs(version - gDigest.getMaxVersion() );
diffDigests.add( new GossipDigest(ep, gDigest.getGeneration(), diffVersion) );
}
@ -935,7 +935,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
*/
for ( int i = size - 1; i >= 0; --i )
{
gDigestList.add( epToDigestMap.get(diffDigests.get(i).getEndPoint()) );
gDigestList.add( epToDigestMap.get(diffDigests.get(i).getEndpoint()) );
}
}
}
@ -957,7 +957,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
{
GossipDigestAckMessage gDigestAckMessage = GossipDigestAckMessage.serializer().deserialize(dis);
List<GossipDigest> gDigestList = gDigestAckMessage.getGossipDigestList();
Map<InetAddress, EndPointState> epStateMap = gDigestAckMessage.getEndPointStateMap();
Map<InetAddress, EndPointState> epStateMap = gDigestAckMessage.getEndpointStateMap();
if ( epStateMap.size() > 0 )
{
@ -970,7 +970,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
Map<InetAddress, EndPointState> deltaEpStateMap = new HashMap<InetAddress, EndPointState>();
for( GossipDigest gDigest : gDigestList )
{
InetAddress addr = gDigest.getEndPoint();
InetAddress addr = gDigest.getEndpoint();
EndPointState localEpStatePtr = Gossiper.instance.getStateForVersionBiggerThan(addr, gDigest.getMaxVersion());
if ( localEpStatePtr != null )
deltaEpStateMap.put(addr, localEpStatePtr);
@ -1010,7 +1010,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
{
throw new RuntimeException(e);
}
Map<InetAddress, EndPointState> remoteEpStateMap = gDigestAck2Message.getEndPointStateMap();
Map<InetAddress, EndPointState> remoteEpStateMap = gDigestAck2Message.getEndpointStateMap();
/* Notify the Failure Detector */
Gossiper.instance.notifyFailureDetector(remoteEpStateMap);
Gossiper.instance.applyStateLocally(remoteEpStateMap);

10
src/java/org/apache/cassandra/hadoop/ColumnFamilySplit.java
View File

@ -84,9 +84,9 @@ public class ColumnFamilySplit extends InputSplit implements Writable
out.writeUTF(endToken);
out.writeInt(dataNodes.length);
for (String endPoint : dataNodes)
for (String endpoint : dataNodes)
{
out.writeUTF(endPoint);
out.writeUTF(endpoint);
}
}
@ -95,9 +95,9 @@ public class ColumnFamilySplit extends InputSplit implements Writable
startToken = in.readUTF();
endToken = in.readUTF();
int numOfEndPoints = in.readInt();
dataNodes = new String[numOfEndPoints];
for(int i = 0; i < numOfEndPoints; i++)
int numOfEndpoints = in.readInt();
dataNodes = new String[numOfEndpoints];
for(int i = 0; i < numOfEndpoints; i++)
{
dataNodes[i] = in.readUTF();
}

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

@ -79,7 +79,7 @@ public abstract class AbstractReplicationStrategy
{
Multimap<InetAddress, InetAddress> map = HashMultimap.create(targets.size(), 1);
IEndPointSnitch endPointSnitch = DatabaseDescriptor.getEndPointSnitch();
IEndPointSnitch endpointSnitch = DatabaseDescriptor.getEndpointSnitch();
// first, add the live endpoints
for (InetAddress ep : targets)
@ -106,7 +106,7 @@ public abstract class AbstractReplicationStrategy
InetAddress destination = map.isEmpty()
? localAddress
: endPointSnitch.getSortedListByProximity(localAddress, map.keySet()).get(0);
: endpointSnitch.getSortedListByProximity(localAddress, map.keySet()).get(0);
map.put(destination, ep);
}

8
src/java/org/apache/cassandra/locator/DatacenterEndPointSnitch.java
View File

@ -75,10 +75,10 @@ public class DatacenterEndPointSnitch extends AbstractEndpointSnitch
/**
* Return the rack for which an endpoint resides in
*/
public String getRackForEndPoint(InetAddress endPoint)
public String getRackForEndpoint(InetAddress endpoint)
throws UnknownHostException
{
byte[] ipQuads = getIPAddress(endPoint.getHostAddress());
byte[] ipQuads = getIPAddress(endpoint.getHostAddress());
return ipRAC.get(ipQuads[1]).get(ipQuads[2]);
}
@ -93,11 +93,11 @@ public class DatacenterEndPointSnitch extends AbstractEndpointSnitch
{
try
{
String[] dcNames = xmlUtils.getNodeValues("/EndPoints/DataCenter/name");
String[] dcNames = xmlUtils.getNodeValues("/Endpoints/DataCenter/name");
for (String dcName : dcNames)
{
// Parse the Datacenter Quaderant.
String dcXPath = "/EndPoints/DataCenter[name='" + dcName + "']";
String dcXPath = "/Endpoints/DataCenter[name='" + dcName + "']";
String dcIPQuad = xmlUtils.getNodeValue(dcXPath + "/ip2ndQuad");
String replicationFactor = xmlUtils.getNodeValue(dcXPath + "/replicationFactor");
byte dcByte = intToByte(Integer.parseInt(dcIPQuad));

32
src/java/org/apache/cassandra/locator/DatacenterShardStategy.java
View File

@ -50,11 +50,11 @@ public class DatacenterShardStategy extends AbstractReplicationStrategy
private static int locQFactor = 0;
ArrayList<Token> tokens;
private List<InetAddress> localEndPoints = new ArrayList<InetAddress>();
private List<InetAddress> localEndpoints = new ArrayList<InetAddress>();
private List<InetAddress> getLocalEndPoints()
private List<InetAddress> getLocalEndpoints()
{
return new ArrayList<InetAddress>(localEndPoints);
return new ArrayList<InetAddress>(localEndpoints);
}
private Map<String, Integer> getQuorumRepFactor()
@ -66,18 +66,18 @@ public class DatacenterShardStategy extends AbstractReplicationStrategy
* This Method will get the required information of the EndPoint from the
* DataCenterEndPointSnitch and poopulates this singleton class.
*/
private synchronized void loadEndPoints(TokenMetadata metadata) throws IOException
private synchronized void loadEndpoints(TokenMetadata metadata) throws IOException
{
this.tokens = new ArrayList<Token>(metadata.sortedTokens());
String localDC = ((DatacenterEndPointSnitch)snitch_).getLocation(InetAddress.getLocalHost());
dcMap = new HashMap<String, List<Token>>();
for (Token token : this.tokens)
{
InetAddress endPoint = metadata.getEndPoint(token);
String dataCenter = ((DatacenterEndPointSnitch)snitch_).getLocation(endPoint);
InetAddress endpoint = metadata.getEndpoint(token);
String dataCenter = ((DatacenterEndPointSnitch)snitch_).getLocation(endpoint);
if (dataCenter.equals(localDC))
{
localEndPoints.add(endPoint);
localEndpoints.add(endpoint);
}
List<Token> lst = dcMap.get(dataCenter);
if (lst == null)
@ -112,7 +112,7 @@ public class DatacenterShardStategy extends AbstractReplicationStrategy
super(tokenMetadata, snitch);
if ((!(snitch instanceof DatacenterEndPointSnitch)))
{
throw new IllegalArgumentException("DatacenterShardStrategy requires DatacenterEndpointSnitch");
throw new IllegalArgumentException("DatacenterShardStrategy requires DatacenterEndPointSnitch");
}
}
@ -137,7 +137,7 @@ public class DatacenterShardStategy extends AbstractReplicationStrategy
if (null == tokens || tokens.size() != metadata.sortedTokens().size())
{
loadEndPoints(metadata);
loadEndpoints(metadata);
}
for (String dc : dcMap.keySet())
@ -149,16 +149,16 @@ public class DatacenterShardStategy extends AbstractReplicationStrategy
boolean doneDataCenterItr;
// Add the node at the index by default
Iterator<Token> iter = TokenMetadata.ringIterator(tokens, searchToken);
InetAddress primaryHost = metadata.getEndPoint(iter.next());
InetAddress primaryHost = metadata.getEndpoint(iter.next());
forloopReturn.add(primaryHost);
while (forloopReturn.size() < replicas_ && iter.hasNext())
{
Token t = iter.next();
InetAddress endPointOfInterest = metadata.getEndPoint(t);
InetAddress endpointOfInterest = metadata.getEndpoint(t);
if (forloopReturn.size() < replicas_ - 1)
{
forloopReturn.add(endPointOfInterest);
forloopReturn.add(endpointOfInterest);
continue;
}
else
@ -169,9 +169,9 @@ public class DatacenterShardStategy extends AbstractReplicationStrategy
// Now try to find one on a different rack
if (!bOtherRack)
{
if (!((DatacenterEndPointSnitch)snitch_).isOnSameRack(primaryHost, endPointOfInterest))
if (!((DatacenterEndPointSnitch)snitch_).isOnSameRack(primaryHost, endpointOfInterest))
{
forloopReturn.add(metadata.getEndPoint(t));
forloopReturn.add(metadata.getEndpoint(t));
bOtherRack = true;
}
}
@ -193,9 +193,9 @@ public class DatacenterShardStategy extends AbstractReplicationStrategy
while (forloopReturn.size() < replicas_ && iter.hasNext())
{
Token t = iter.next();
if (!forloopReturn.contains(metadata.getEndPoint(t)))
if (!forloopReturn.contains(metadata.getEndpoint(t)))
{
forloopReturn.add(metadata.getEndPoint(t));
forloopReturn.add(metadata.getEndpoint(t));
}
}
}

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

@ -56,7 +56,7 @@ public class RackAwareStrategy extends AbstractReplicationStrategy
Iterator<Token> iter = TokenMetadata.ringIterator(tokens, token);
Token primaryToken = iter.next();
endpoints.add(metadata.getEndPoint(primaryToken));
endpoints.add(metadata.getEndpoint(primaryToken));
boolean bDataCenter = false;
boolean bOtherRack = false;
@ -66,24 +66,24 @@ public class RackAwareStrategy extends AbstractReplicationStrategy
{
// First try to find one in a different data center
Token t = iter.next();
if (!((EndPointSnitch)snitch_).isInSameDataCenter(metadata.getEndPoint(primaryToken), metadata.getEndPoint(t)))
if (!((EndPointSnitch)snitch_).isInSameDataCenter(metadata.getEndpoint(primaryToken), metadata.getEndpoint(t)))
{
// If we have already found something in a diff datacenter no need to find another
if (!bDataCenter)
{
endpoints.add(metadata.getEndPoint(t));
endpoints.add(metadata.getEndpoint(t));
bDataCenter = true;
}
continue;
}
// Now try to find one on a different rack
if (!((EndPointSnitch)snitch_).isOnSameRack(metadata.getEndPoint(primaryToken), metadata.getEndPoint(t)) &&
((EndPointSnitch)snitch_).isInSameDataCenter(metadata.getEndPoint(primaryToken), metadata.getEndPoint(t)))
if (!((EndPointSnitch)snitch_).isOnSameRack(metadata.getEndpoint(primaryToken), metadata.getEndpoint(t)) &&
((EndPointSnitch)snitch_).isInSameDataCenter(metadata.getEndpoint(primaryToken), metadata.getEndpoint(t)))
{
// If we have already found something in a diff rack no need to find another
if (!bOtherRack)
{
endpoints.add(metadata.getEndPoint(t));
endpoints.add(metadata.getEndpoint(t));
bOtherRack = true;
}
}
@ -103,8 +103,8 @@ public class RackAwareStrategy extends AbstractReplicationStrategy
while (endpoints.size() < replicas && iter.hasNext())
{
Token t = iter.next();
if (!endpoints.contains(metadata.getEndPoint(t)))
endpoints.add(metadata.getEndPoint(t));
if (!endpoints.contains(metadata.getEndpoint(t)))
endpoints.add(metadata.getEndpoint(t));
}
}

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

@ -53,7 +53,7 @@ public class RackUnawareStrategy extends AbstractReplicationStrategy
Iterator<Token> iter = TokenMetadata.ringIterator(tokens, token);
while (endpoints.size() < replicas && iter.hasNext())
{
endpoints.add(metadata.getEndPoint(iter.next()));
endpoints.add(metadata.getEndpoint(iter.next()));
}
return endpoints;

102
src/java/org/apache/cassandra/locator/TokenMetadata.java
View File

@ -35,7 +35,7 @@ import org.apache.commons.lang.StringUtils;
public class TokenMetadata
{
/* Maintains token to endpoint map of every node in the cluster. */
private BiMap<Token, InetAddress> tokenToEndPointMap;
private BiMap<Token, InetAddress> tokenToEndpointMap;
// Suppose that there is a ring of nodes A, C and E, with replication factor 3.
// Node D bootstraps between C and E, so its pending ranges will be E-A, A-C and C-D.
@ -51,7 +51,7 @@ public class TokenMetadata
// An anonymous pending ranges list is not enough, as that does not tell which node is leaving
// and/or if the ranges are there because of bootstrap or leave operation.
// (See CASSANDRA-603 for more detail + examples).
private Set<InetAddress> leavingEndPoints;
private Set<InetAddress> leavingEndpoints;
private ConcurrentMap<String, Multimap<Range, InetAddress>> pendingRanges;
@ -64,20 +64,20 @@ public class TokenMetadata
this(null);
}
public TokenMetadata(BiMap<Token, InetAddress> tokenToEndPointMap)
public TokenMetadata(BiMap<Token, InetAddress> tokenToEndpointMap)
{
if (tokenToEndPointMap == null)
tokenToEndPointMap = HashBiMap.create();
this.tokenToEndPointMap = tokenToEndPointMap;
if (tokenToEndpointMap == null)
tokenToEndpointMap = HashBiMap.create();
this.tokenToEndpointMap = tokenToEndpointMap;
bootstrapTokens = HashBiMap.create();
leavingEndPoints = new HashSet<InetAddress>();
leavingEndpoints = new HashSet<InetAddress>();
pendingRanges = new ConcurrentHashMap<String, Multimap<Range, InetAddress>>();
sortedTokens = sortTokens();
}
private List<Token> sortTokens()
{
List<Token> tokens = new ArrayList<Token>(tokenToEndPointMap.keySet());
List<Token> tokens = new ArrayList<Token>(tokenToEndpointMap.keySet());
Collections.sort(tokens);
return Collections.unmodifiableList(tokens);
}
@ -102,12 +102,12 @@ public class TokenMetadata
try
{
bootstrapTokens.inverse().remove(endpoint);
tokenToEndPointMap.inverse().remove(endpoint);
if (!endpoint.equals(tokenToEndPointMap.put(token, endpoint)))
tokenToEndpointMap.inverse().remove(endpoint);
if (!endpoint.equals(tokenToEndpointMap.put(token, endpoint)))
{
sortedTokens = sortTokens();
}
leavingEndPoints.remove(endpoint);
leavingEndpoints.remove(endpoint);
}
finally
{
@ -123,15 +123,15 @@ public class TokenMetadata
lock.writeLock().lock();
try
{
InetAddress oldEndPoint = null;
InetAddress oldEndpoint = null;
oldEndPoint = bootstrapTokens.get(token);
if (oldEndPoint != null && !oldEndPoint.equals(endpoint))
throw new RuntimeException("Bootstrap Token collision between " + oldEndPoint + " and " + endpoint + " (token " + token);
oldEndpoint = bootstrapTokens.get(token);
if (oldEndpoint != null && !oldEndpoint.equals(endpoint))
throw new RuntimeException("Bootstrap Token collision between " + oldEndpoint + " and " + endpoint + " (token " + token);
oldEndPoint = tokenToEndPointMap.get(token);
if (oldEndPoint != null && !oldEndPoint.equals(endpoint))
throw new RuntimeException("Bootstrap Token collision between " + oldEndPoint + " and " + endpoint + " (token " + token);
oldEndpoint = tokenToEndpointMap.get(token);
if (oldEndpoint != null && !oldEndpoint.equals(endpoint))
throw new RuntimeException("Bootstrap Token collision between " + oldEndpoint + " and " + endpoint + " (token " + token);
bootstrapTokens.inverse().remove(endpoint);
bootstrapTokens.put(token, endpoint);
@ -157,14 +157,14 @@ public class TokenMetadata
}
}
public void addLeavingEndPoint(InetAddress endpoint)
public void addLeavingEndpoint(InetAddress endpoint)
{
assert endpoint != null;
lock.writeLock().lock();
try
{
leavingEndPoints.add(endpoint);
leavingEndpoints.add(endpoint);
}
finally
{
@ -172,14 +172,14 @@ public class TokenMetadata
}
}
public void removeLeavingEndPoint(InetAddress endpoint)
public void removeLeavingEndpoint(InetAddress endpoint)
{
assert endpoint != null;
lock.writeLock().lock();
try
{
leavingEndPoints.remove(endpoint);
leavingEndpoints.remove(endpoint);
}
finally
{
@ -189,13 +189,13 @@ public class TokenMetadata
public void removeEndpoint(InetAddress endpoint)
{
assert tokenToEndPointMap.containsValue(endpoint);
assert tokenToEndpointMap.containsValue(endpoint);
lock.writeLock().lock();
try
{
bootstrapTokens.inverse().remove(endpoint);
tokenToEndPointMap.inverse().remove(endpoint);
leavingEndPoints.remove(endpoint);
tokenToEndpointMap.inverse().remove(endpoint);
leavingEndpoints.remove(endpoint);
sortedTokens = sortTokens();
}
finally
@ -212,7 +212,7 @@ public class TokenMetadata
lock.readLock().lock();
try
{
return tokenToEndPointMap.inverse().get(endpoint);
return tokenToEndpointMap.inverse().get(endpoint);
}
finally
{
@ -227,7 +227,7 @@ public class TokenMetadata
lock.readLock().lock();
try
{
return tokenToEndPointMap.inverse().containsKey(endpoint);
return tokenToEndpointMap.inverse().containsKey(endpoint);
}
finally
{
@ -242,7 +242,7 @@ public class TokenMetadata
lock.readLock().lock();
try
{
return leavingEndPoints.contains(endpoint);
return leavingEndpoints.contains(endpoint);
}
finally
{
@ -252,12 +252,12 @@ public class TokenMetadata
public InetAddress getFirstEndpoint()
{
assert tokenToEndPointMap.size() > 0;
assert tokenToEndpointMap.size() > 0;
lock.readLock().lock();
try
{
return tokenToEndPointMap.get(sortedTokens.get(0));
return tokenToEndpointMap.get(sortedTokens.get(0));
}
finally
{
@ -266,7 +266,7 @@ public class TokenMetadata
}
/**
* Create a copy of TokenMetadata with only tokenToEndPointMap. That is, pending ranges,
* Create a copy of TokenMetadata with only tokenToEndpointMap. That is, pending ranges,
* bootstrap tokens and leaving endpoints are not included in the copy.
*/
public TokenMetadata cloneOnlyTokenMap()
@ -274,7 +274,7 @@ public class TokenMetadata
lock.readLock().lock();
try
{
return new TokenMetadata(HashBiMap.create(tokenToEndPointMap));
return new TokenMetadata(HashBiMap.create(tokenToEndpointMap));
}
finally
{
@ -283,7 +283,7 @@ public class TokenMetadata
}
/**
* Create a copy of TokenMetadata with tokenToEndPointMap reflecting situation after all
* Create a copy of TokenMetadata with tokenToEndpointMap reflecting situation after all
* current leave operations have finished.
*/
public TokenMetadata cloneAfterAllLeft()
@ -292,8 +292,8 @@ public class TokenMetadata
try
{
TokenMetadata allLeftMetadata = cloneOnlyTokenMap();
for (InetAddress endPoint : leavingEndPoints)
allLeftMetadata.removeEndpoint(endPoint);
for (InetAddress endpoint : leavingEndpoints)
allLeftMetadata.removeEndpoint(endpoint);
return allLeftMetadata;
}
finally
@ -302,12 +302,12 @@ public class TokenMetadata
}
}
public InetAddress getEndPoint(Token token)
public InetAddress getEndpoint(Token token)
{
lock.readLock().lock();
try
{
return tokenToEndPointMap.get(token);
return tokenToEndpointMap.get(token);
}
finally
{
@ -372,7 +372,7 @@ public class TokenMetadata
{
List tokens = sortedTokens();
int index = Collections.binarySearch(tokens, token);
assert index >= 0 : token + " not found in " + StringUtils.join(tokenToEndPointMap.keySet(), ", ");
assert index >= 0 : token + " not found in " + StringUtils.join(tokenToEndpointMap.keySet(), ", ");
return (Token) (index == 0 ? tokens.get(tokens.size() - 1) : tokens.get(index - 1));
}
@ -380,13 +380,13 @@ public class TokenMetadata
{
List tokens = sortedTokens();
int index = Collections.binarySearch(tokens, token);
assert index >= 0 : token + " not found in " + StringUtils.join(tokenToEndPointMap.keySet(), ", ");
assert index >= 0 : token + " not found in " + StringUtils.join(tokenToEndpointMap.keySet(), ", ");
return (Token) ((index == (tokens.size() - 1)) ? tokens.get(0) : tokens.get(index + 1));
}
public InetAddress getSuccessor(InetAddress endPoint)
public InetAddress getSuccessor(InetAddress endpoint)
{
return getEndPoint(getSuccessor(getToken(endPoint)));
return getEndpoint(getSuccessor(getToken(endpoint)));
}
/** caller should not modify bootstrapTokens */
@ -395,10 +395,10 @@ public class TokenMetadata
return bootstrapTokens;
}
/** caller should not modify leavigEndPoints */
public Set<InetAddress> getLeavingEndPoints()
/** caller should not modify leavigEndpoints */
public Set<InetAddress> getLeavingEndpoints()
{
return leavingEndPoints;
return leavingEndpoints;
}
/**
@ -443,8 +443,8 @@ public class TokenMetadata
public void clearUnsafe()
{
bootstrapTokens.clear();
tokenToEndPointMap.clear();
leavingEndPoints.clear();
tokenToEndpointMap.clear();
leavingEndpoints.clear();
pendingRanges.clear();
}
@ -454,7 +454,7 @@ public class TokenMetadata
lock.readLock().lock();
try
{
Set<InetAddress> eps = tokenToEndPointMap.inverse().keySet();
Set<InetAddress> eps = tokenToEndpointMap.inverse().keySet();
if (!eps.isEmpty())
{
@ -464,7 +464,7 @@ public class TokenMetadata
{
sb.append(ep);
sb.append(":");
sb.append(tokenToEndPointMap.inverse().get(ep));
sb.append(tokenToEndpointMap.inverse().get(ep));
sb.append(System.getProperty("line.separator"));
}
}
@ -480,11 +480,11 @@ public class TokenMetadata
}
}
if (!leavingEndPoints.isEmpty())
if (!leavingEndpoints.isEmpty())
{
sb.append("Leaving EndPoints:");
sb.append("Leaving Endpoints:");
sb.append(System.getProperty("line.separator"));
for (InetAddress ep : leavingEndPoints)
for (InetAddress ep : leavingEndpoints)
{
sb.append(ep);
sb.append(System.getProperty("line.separator"));

4
src/java/org/apache/cassandra/net/CompactEndPointSerializationHelper.java
View File

@ -23,9 +23,9 @@ import java.net.InetAddress;
public class CompactEndPointSerializationHelper
{
public static void serialize(InetAddress endPoint, DataOutputStream dos) throws IOException
public static void serialize(InetAddress endpoint, DataOutputStream dos) throws IOException
{
byte[] buf = endPoint.getAddress();
byte[] buf = endpoint.getAddress();
dos.writeByte(buf.length);
dos.write(buf);
}

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

@ -559,8 +559,8 @@ public class AntiEntropyService
rtree.partitioner(StorageService.getPartitioner());
// determine the ranges where responsibility overlaps
Set<Range> interesting = new HashSet(ss.getRangesForEndPoint(cf.left, local));
interesting.retainAll(ss.getRangesForEndPoint(cf.left, remote));
Set<Range> interesting = new HashSet(ss.getRangesForEndpoint(cf.left, local));
interesting.retainAll(ss.getRangesForEndpoint(cf.left, remote));
// compare trees, and filter out uninteresting differences
for (MerkleTree.TreeRange diff : MerkleTree.difference(ltree, rtree))

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

@ -42,14 +42,14 @@ public class DatacenterSyncWriteResponseHandler extends WriteResponseHandler
{
private final Map<String, Integer> dcResponses = new HashMap<String, Integer>();
private final Map<String, Integer> responseCounts;
private final DatacenterEndPointSnitch endPointSnitch;
private final DatacenterEndPointSnitch endpointSnitch;
public DatacenterSyncWriteResponseHandler(Map<String, Integer> responseCounts, String table)
{
// Response is been managed by the map so make it 1 for the superclass.
super(1, table);
this.responseCounts = responseCounts;
endPointSnitch = (DatacenterEndPointSnitch) DatabaseDescriptor.getEndPointSnitch();
endpointSnitch = (DatacenterEndPointSnitch) DatabaseDescriptor.getEndpointSnitch();
}
@Override
@ -60,7 +60,7 @@ public class DatacenterSyncWriteResponseHandler extends WriteResponseHandler
{
try
{
String dataCenter = endPointSnitch.getLocation(message.getFrom());
String dataCenter = endpointSnitch.getLocation(message.getFrom());
Object blockFor = responseCounts.get(dataCenter);
// If this DC needs to be blocked then do the below.
if (blockFor != null)

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

@ -50,7 +50,7 @@ public class DatacenterWriteResponseHandler extends WriteResponseHandler
// Response is been managed by the map so the waitlist size really doesnt matter.
super(blockFor, table);
this.blockFor = new AtomicInteger(blockFor);
endpointsnitch = (DatacenterEndPointSnitch) DatabaseDescriptor.getEndPointSnitch();
endpointsnitch = (DatacenterEndPointSnitch) DatabaseDescriptor.getEndpointSnitch();
localEndpoint = FBUtilities.getLocalAddress();
}

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

@ -62,7 +62,7 @@ public class MigrationManager implements IEndPointStateChangeSubscriber
}
/** gets called after a this node joins a cluster */
public void onAlive(InetAddress endpoint, EndPointState state)
public void onAlive(InetAddress endpoint, EndPointState state)
{
ApplicationState appState = state.getApplicationState(MIGRATION_STATE);
if (appState != null)

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

@ -65,7 +65,7 @@ public class ReadResponseResolver implements IResponseResolver<Row>
{
long startTime = System.currentTimeMillis();
List<ColumnFamily> versions = new ArrayList<ColumnFamily>();
List<InetAddress> endPoints = new ArrayList<InetAddress>();
List<InetAddress> endpoints = new ArrayList<InetAddress>();
DecoratedKey key = null;
byte[] digest = new byte[0];
boolean isDigestQuery = false;
@ -89,7 +89,7 @@ public class ReadResponseResolver implements IResponseResolver<Row>
else
{
versions.add(result.row().cf);
endPoints.add(response.getFrom());
endpoints.add(response.getFrom());
key = result.row().key;
}
}
@ -109,7 +109,7 @@ public class ReadResponseResolver implements IResponseResolver<Row>
}
ColumnFamily resolved = resolveSuperset(versions);
maybeScheduleRepairs(resolved, table, key, versions, endPoints);
maybeScheduleRepairs(resolved, table, key, versions, endpoints);
if (logger_.isDebugEnabled())
logger_.debug("resolve: " + (System.currentTimeMillis() - startTime) + " ms.");
@ -120,7 +120,7 @@ public class ReadResponseResolver implements IResponseResolver<Row>
* For each row version, compare with resolved (the superset of all row versions);
* if it is missing anything, send a mutation to the endpoint it come from.
*/
public static void maybeScheduleRepairs(ColumnFamily resolved, String table, DecoratedKey key, List<ColumnFamily> versions, List<InetAddress> endPoints)
public static void maybeScheduleRepairs(ColumnFamily resolved, String table, DecoratedKey key, List<ColumnFamily> versions, List<InetAddress> endpoints)
{
for (int i = 0; i < versions.size(); i++)
{
@ -132,7 +132,7 @@ public class ReadResponseResolver implements IResponseResolver<Row>
RowMutation rowMutation = new RowMutation(table, key.key);
rowMutation.add(diffCf);
RowMutationMessage rowMutationMessage = new RowMutationMessage(rowMutation);
ReadRepairManager.instance.schedule(endPoints.get(i), rowMutationMessage);
ReadRepairManager.instance.schedule(endpoints.get(i), rowMutationMessage);
}
}

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

@ -69,14 +69,14 @@ public class StorageLoadBalancer implements IEndPointStateChangeSubscriber
/*
int threshold = (int)(StorageLoadBalancer.TOPHEAVY_RATIO * averageSystemLoad());
int myLoad = localLoad();
InetAddress predecessor = StorageService.instance.getPredecessor(StorageService.getLocalStorageEndPoint());
InetAddress predecessor = StorageService.instance.getPredecessor(StorageService.getLocalStorageEndpoint());
if (logger_.isDebugEnabled())
logger_.debug("Trying to relocate the predecessor {}", predecessor);
boolean value = tryThisNode(myLoad, threshold, predecessor);
if ( !value )
{
loadInfo2_.remove(predecessor);
InetAddress successor = StorageService.instance.getSuccessor(StorageService.getLocalStorageEndPoint());
InetAddress successor = StorageService.instance.getSuccessor(StorageService.getLocalStorageEndpoint());
if (logger_.isDebugEnabled())
logger_.debug("Trying to relocate the successor {}", successor);
value = tryThisNode(myLoad, threshold, successor);
@ -136,7 +136,7 @@ public class StorageLoadBalancer implements IEndPointStateChangeSubscriber
BigInteger targetToken = BootstrapAndLbHelper.getTokenBasedOnPrimaryCount(keyCount);
// Send a MoveMessage and see if this node is relocateable
MoveMessage moveMessage = new MoveMessage(targetToken);
Message message = new Message(StorageService.getLocalStorageEndPoint(), StorageLoadBalancer.lbStage_, StorageLoadBalancer.moveMessageVerbHandler_, new Object[]{moveMessage});
Message message = new Message(StorageService.getLocalStorageEndpoint(), StorageLoadBalancer.lbStage_, StorageLoadBalancer.moveMessageVerbHandler_, new Object[]{moveMessage});
if (logger_.isDebugEnabled())
logger_.debug("Sending a move message to {}", target);
IAsyncResult result = MessagingService.getMessagingInstance().sendRR(message, target);
@ -205,7 +205,7 @@ public class StorageLoadBalancer implements IEndPointStateChangeSubscriber
if ( isHeavyNode() )
{
if (logger_.isDebugEnabled())
logger_.debug(StorageService.getLocalStorageEndPoint() + " is a heavy node with load " + localLoad());
logger_.debug(StorageService.getLocalStorageEndpoint() + " is a heavy node with load " + localLoad());
// lb_.schedule( new LoadBalancer(), StorageLoadBalancer.delay_, TimeUnit.MINUTES );
}
*/
@ -230,7 +230,7 @@ public class StorageLoadBalancer implements IEndPointStateChangeSubscriber
if ( !isMoveable_.get() )
return false;
int myload = localLoad();
InetAddress successor = StorageService.instance.getSuccessor(StorageService.getLocalStorageEndPoint());
InetAddress successor = StorageService.instance.getSuccessor(StorageService.getLocalStorageEndpoint());
LoadInfo li = loadInfo2_.get(successor);
// "load" is NULL means that the successor node has not
// yet gossiped its load information. We should return

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

@ -345,14 +345,14 @@ public class StorageProxy implements StorageProxyMBean
for (ReadCommand command: commands)
{
InetAddress endPoint = StorageService.instance.findSuitableEndPoint(command.table, command.key);
InetAddress endpoint = StorageService.instance.findSuitableEndpoint(command.table, command.key);
Message message = command.makeReadMessage();
if (logger.isDebugEnabled())
logger.debug("weakreadremote reading " + command + " from " + message.getMessageId() + "@" + endPoint);
logger.debug("weakreadremote reading " + command + " from " + message.getMessageId() + "@" + endpoint);
if (randomlyReadRepair(command))
message.setHeader(ReadCommand.DO_REPAIR, ReadCommand.DO_REPAIR.getBytes());
iars.add(MessagingService.instance.sendRR(message, endPoint));
iars.add(MessagingService.instance.sendRR(message, endpoint));
}
for (IAsyncResult iar: iars)
@ -429,7 +429,7 @@ public class StorageProxy implements StorageProxyMBean
private static List<Row> strongRead(List<ReadCommand> commands, ConsistencyLevel consistency_level) throws IOException, UnavailableException, TimeoutException
{
List<QuorumResponseHandler<Row>> quorumResponseHandlers = new ArrayList<QuorumResponseHandler<Row>>();
List<InetAddress[]> commandEndPoints = new ArrayList<InetAddress[]>();
List<InetAddress[]> commandEndpoints = new ArrayList<InetAddress[]>();
List<Row> rows = new ArrayList<Row>();
int commandIndex = 0;
@ -442,14 +442,14 @@ public class StorageProxy implements StorageProxyMBean
Message message = command.makeReadMessage();
Message messageDigestOnly = readMessageDigestOnly.makeReadMessage();
InetAddress dataPoint = StorageService.instance.findSuitableEndPoint(command.table, command.key);
InetAddress dataPoint = StorageService.instance.findSuitableEndpoint(command.table, command.key);
List<InetAddress> endpointList = StorageService.instance.getLiveNaturalEndpoints(command.table, command.key);
final String table = command.table;
int responseCount = determineBlockFor(DatabaseDescriptor.getReplicationFactor(table), consistency_level);
if (endpointList.size() < responseCount)
throw new UnavailableException();
InetAddress[] endPoints = new InetAddress[endpointList.size()];
InetAddress[] endpoints = new InetAddress[endpointList.size()];
Message messages[] = new Message[endpointList.size()];
// data-request message is sent to dataPoint, the node that will actually get
// the data for us. The other replicas are only sent a digest query.
@ -457,15 +457,15 @@ public class StorageProxy implements StorageProxyMBean
for (InetAddress endpoint : endpointList)
{
Message m = endpoint.equals(dataPoint) ? message : messageDigestOnly;
endPoints[n] = endpoint;
endpoints[n] = endpoint;
messages[n++] = m;
if (logger.isDebugEnabled())
logger.debug("strongread reading " + (m == message ? "data" : "digest") + " for " + command + " from " + m.getMessageId() + "@" + endpoint);
}
QuorumResponseHandler<Row> quorumResponseHandler = new QuorumResponseHandler<Row>(DatabaseDescriptor.getQuorum(command.table), new ReadResponseResolver(command.table, responseCount));
MessagingService.instance.sendRR(messages, endPoints, quorumResponseHandler);
MessagingService.instance.sendRR(messages, endpoints, quorumResponseHandler);
quorumResponseHandlers.add(quorumResponseHandler);
commandEndPoints.add(endPoints);
commandEndpoints.add(endpoints);
}
for (QuorumResponseHandler<Row> quorumResponseHandler: quorumResponseHandlers)
@ -492,7 +492,7 @@ public class StorageProxy implements StorageProxyMBean
readResponseResolverRepair);
logger.info("DigestMismatchException: " + ex.getMessage());
Message messageRepair = command.makeReadMessage();
MessagingService.instance.sendRR(messageRepair, commandEndPoints.get(commandIndex), quorumResponseHandlerRepair);
MessagingService.instance.sendRR(messageRepair, commandEndpoints.get(commandIndex), quorumResponseHandlerRepair);
try
{
row = quorumResponseHandlerRepair.get();
@ -675,7 +675,7 @@ public class StorageProxy implements StorageProxyMBean
if (endpoints.size() < responseCount)
throw new UnavailableException();
DatabaseDescriptor.getEndPointSnitch().sortByProximity(FBUtilities.getLocalAddress(), endpoints);
DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), endpoints);
List<InetAddress> endpointsForCL = endpoints.subList(0, responseCount);
Set<AbstractBounds> restrictedRanges = queryRange.restrictTo(nodeRange);
for (AbstractBounds range : restrictedRanges)

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

@ -125,12 +125,12 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
public Collection<Range> getLocalRanges(String table)
{
return getRangesForEndPoint(table, FBUtilities.getLocalAddress());
return getRangesForEndpoint(table, FBUtilities.getLocalAddress());
}
public Range getLocalPrimaryRange()
{
return getPrimaryRangeForEndPoint(FBUtilities.getLocalAddress());
return getPrimaryRangeForEndpoint(FBUtilities.getLocalAddress());
}
/* This abstraction maintains the token/endpoint metadata information */
@ -274,7 +274,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
try
{
Constructor<? extends AbstractReplicationStrategy> constructor = cls.getConstructor(parameterTypes);
replicationStrategy = constructor.newInstance(tokenMetadata, DatabaseDescriptor.getEndPointSnitch());
replicationStrategy = constructor.newInstance(tokenMetadata, DatabaseDescriptor.getEndpointSnitch());
}
catch (Exception e)
{
@ -451,7 +451,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
* @param keyspace
* @return
*/
public Map<Range, List<String>> getRangeToEndPointMap(String keyspace)
public Map<Range, List<String>> getRangeToEndpointMap(String keyspace)
{
// some people just want to get a visual representation of things. Allow null and set it to the first
// non-system table.
@ -470,7 +470,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
public Map<Range, List<InetAddress>> getRangeToAddressMap(String keyspace)
{
List<Range> ranges = getAllRanges(tokenMetadata_.sortedTokens());
return constructRangeToEndPointMap(keyspace, ranges);
return constructRangeToEndpointMap(keyspace, ranges);
}
/**
@ -479,14 +479,14 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
* @param ranges
* @return mapping of ranges to the replicas responsible for them.
*/
private Map<Range, List<InetAddress>> constructRangeToEndPointMap(String keyspace, List<Range> ranges)
private Map<Range, List<InetAddress>> constructRangeToEndpointMap(String keyspace, List<Range> ranges)
{
Map<Range, List<InetAddress>> rangeToEndPointMap = new HashMap<Range, List<InetAddress>>();
Map<Range, List<InetAddress>> rangeToEndpointMap = new HashMap<Range, List<InetAddress>>();
for (Range range : ranges)
{
rangeToEndPointMap.put(range, getReplicationStrategy(keyspace).getNaturalEndpoints(range.right, keyspace));
rangeToEndpointMap.put(range, getReplicationStrategy(keyspace).getNaturalEndpoints(range.right, keyspace));
}
return rangeToEndPointMap;
return rangeToEndpointMap;
}
/*
@ -521,32 +521,32 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
/**
* Handle node bootstrap
*
* @param endPoint bootstrapping node
* @param endpoint bootstrapping node
* @param moveValue bootstrap token as string
*/
private void handleStateBootstrap(InetAddress endPoint, String moveValue)
private void handleStateBootstrap(InetAddress endpoint, String moveValue)
{
Token token = getPartitioner().getTokenFactory().fromString(moveValue);
if (logger_.isDebugEnabled())
logger_.debug("Node " + endPoint + " state bootstrapping, token " + token);
logger_.debug("Node " + endpoint + " state bootstrapping, token " + token);
// if this node is present in token metadata, either we have missed intermediate states
// or the node had crashed. Print warning if needed, clear obsolete stuff and
// continue.
if (tokenMetadata_.isMember(endPoint))
if (tokenMetadata_.isMember(endpoint))
{
// If isLeaving is false, we have missed both LEAVING and LEFT. However, if
// isLeaving is true, we have only missed LEFT. Waiting time between completing
// leave operation and rebootstrapping is relatively short, so the latter is quite
// common (not enough time for gossip to spread). Therefore we report only the
// former in the log.
if (!tokenMetadata_.isLeaving(endPoint))
logger_.info("Node " + endPoint + " state jump to bootstrap");
tokenMetadata_.removeEndpoint(endPoint);
if (!tokenMetadata_.isLeaving(endpoint))
logger_.info("Node " + endpoint + " state jump to bootstrap");
tokenMetadata_.removeEndpoint(endpoint);
}
tokenMetadata_.addBootstrapToken(token, endPoint);
tokenMetadata_.addBootstrapToken(token, endpoint);
calculatePendingRanges();
}
@ -554,55 +554,55 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
* Handle node move to normal state. That is, node is entering token ring and participating
* in reads.
*
* @param endPoint node
* @param endpoint node
* @param moveValue token as string
*/
private void handleStateNormal(InetAddress endPoint, String moveValue)
private void handleStateNormal(InetAddress endpoint, String moveValue)
{
Token token = getPartitioner().getTokenFactory().fromString(moveValue);
if (logger_.isDebugEnabled())
logger_.debug("Node " + endPoint + " state normal, token " + token);
logger_.debug("Node " + endpoint + " state normal, token " + token);
if (tokenMetadata_.isMember(endPoint))
logger_.info("Node " + endPoint + " state jump to normal");
if (tokenMetadata_.isMember(endpoint))
logger_.info("Node " + endpoint + " state jump to normal");
tokenMetadata_.updateNormalToken(token, endPoint);
tokenMetadata_.updateNormalToken(token, endpoint);
calculatePendingRanges();
if (!isClientMode)
SystemTable.updateToken(endPoint, token);
SystemTable.updateToken(endpoint, token);
}
/**
* Handle node preparing to leave the ring
*
* @param endPoint node
* @param endpoint node
* @param moveValue token as string
*/
private void handleStateLeaving(InetAddress endPoint, String moveValue)
private void handleStateLeaving(InetAddress endpoint, String moveValue)
{
Token token = getPartitioner().getTokenFactory().fromString(moveValue);
if (logger_.isDebugEnabled())
logger_.debug("Node " + endPoint + " state leaving, token " + token);
logger_.debug("Node " + endpoint + " state leaving, token " + token);
// If the node is previously unknown or tokens do not match, update tokenmetadata to
// have this node as 'normal' (it must have been using this token before the
// leave). This way we'll get pending ranges right.
if (!tokenMetadata_.isMember(endPoint))
if (!tokenMetadata_.isMember(endpoint))
{
logger_.info("Node " + endPoint + " state jump to leaving");
tokenMetadata_.updateNormalToken(token, endPoint);
logger_.info("Node " + endpoint + " state jump to leaving");
tokenMetadata_.updateNormalToken(token, endpoint);
}
else if (!tokenMetadata_.getToken(endPoint).equals(token))
else if (!tokenMetadata_.getToken(endpoint).equals(token))
{
logger_.warn("Node " + endPoint + " 'leaving' token mismatch. Long network partition?");
tokenMetadata_.updateNormalToken(token, endPoint);
logger_.warn("Node " + endpoint + " 'leaving' token mismatch. Long network partition?");
tokenMetadata_.updateNormalToken(token, endpoint);
}
// at this point the endpoint is certainly a member with this token, so let's proceed
// normally
tokenMetadata_.addLeavingEndPoint(endPoint);
tokenMetadata_.addLeavingEndpoint(endpoint);
calculatePendingRanges();
}
@ -610,51 +610,51 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
* Handle node leaving the ring. This can be either because the node was removed manually by
* removetoken command or because of decommission or loadbalance
*
* @param endPoint If reason for leaving is decommission or loadbalance (LEFT_NORMALLY),
* endPoint is the leaving node. If reason manual removetoken (REMOVE_TOKEN), endPoint
* @param endpoint If reason for leaving is decommission or loadbalance (LEFT_NORMALLY),
* endpoint is the leaving node. If reason manual removetoken (REMOVE_TOKEN), endpoint
* parameter is ignored and the operation is based on the token inside moveValue.
* @param moveValue (REMOVE_TOKEN|LEFT_NORMALLY)<Delimiter><token>
*/
private void handleStateLeft(InetAddress endPoint, String moveValue)
private void handleStateLeft(InetAddress endpoint, String moveValue)
{
int index = moveValue.indexOf(Delimiter);
assert (index != -1);
String typeOfState = moveValue.substring(0, index);
Token token = getPartitioner().getTokenFactory().fromString(moveValue.substring(index + 1));
// endPoint itself is leaving
// endpoint itself is leaving
if (typeOfState.equals(LEFT_NORMALLY))
{
if (logger_.isDebugEnabled())
logger_.debug("Node " + endPoint + " state left, token " + token);
logger_.debug("Node " + endpoint + " state left, token " + token);
// If the node is member, remove all references to it. If not, call
// removeBootstrapToken just in case it is there (very unlikely chain of events)
if (tokenMetadata_.isMember(endPoint))
if (tokenMetadata_.isMember(endpoint))
{
if (!tokenMetadata_.getToken(endPoint).equals(token))
logger_.warn("Node " + endPoint + " 'left' token mismatch. Long network partition?");
tokenMetadata_.removeEndpoint(endPoint);
if (!tokenMetadata_.getToken(endpoint).equals(token))
logger_.warn("Node " + endpoint + " 'left' token mismatch. Long network partition?");
tokenMetadata_.removeEndpoint(endpoint);
}
}
else
{
// if we're here, endPoint is not leaving but broadcasting remove token command
// if we're here, endpoint is not leaving but broadcasting remove token command
assert (typeOfState.equals(REMOVE_TOKEN));
InetAddress endPointThatLeft = tokenMetadata_.getEndPoint(token);
InetAddress endpointThatLeft = tokenMetadata_.getEndpoint(token);
// let's make sure that we're not removing ourselves. This can happen when a node
// enters ring as a replacement for a removed node. removeToken for the old node is
// still in gossip, so we will see it.
if (FBUtilities.getLocalAddress().equals(endPointThatLeft))
if (FBUtilities.getLocalAddress().equals(endpointThatLeft))
{
logger_.info("Received removeToken gossip about myself. Is this node a replacement for a removed one?");
return;
}
if (logger_.isDebugEnabled())
logger_.debug("Token " + token + " removed manually (endpoint was " + ((endPointThatLeft == null) ? "unknown" : endPointThatLeft) + ")");
if (endPointThatLeft != null)
logger_.debug("Token " + token + " removed manually (endpoint was " + ((endpointThatLeft == null) ? "unknown" : endpointThatLeft) + ")");
if (endpointThatLeft != null)
{
removeEndPointLocally(endPointThatLeft);
removeEndpointLocally(endpointThatLeft);
}
}
@ -664,14 +664,14 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
}
/**
* endPoint was completely removed from ring (as a result of removetoken command). Remove it
* endpoint was completely removed from ring (as a result of removetoken command). Remove it
* from token metadata and gossip and restore replica count.
*/
private void removeEndPointLocally(InetAddress endPoint)
private void removeEndpointLocally(InetAddress endpoint)
{
restoreReplicaCount(endPoint);
Gossiper.instance.removeEndPoint(endPoint);
tokenMetadata_.removeEndpoint(endPoint);
restoreReplicaCount(endpoint);
Gossiper.instance.removeEndpoint(endpoint);
tokenMetadata_.removeEndpoint(endpoint);
}
/**
@ -709,9 +709,9 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
TokenMetadata tm = StorageService.instance.getTokenMetadata();
Multimap<Range, InetAddress> pendingRanges = HashMultimap.create();
Map<Token, InetAddress> bootstrapTokens = tm.getBootstrapTokens();
Set<InetAddress> leavingEndPoints = tm.getLeavingEndPoints();
Set<InetAddress> leavingEndpoints = tm.getLeavingEndpoints();
if (bootstrapTokens.isEmpty() && leavingEndPoints.isEmpty())
if (bootstrapTokens.isEmpty() && leavingEndpoints.isEmpty())
{
if (logger_.isDebugEnabled())
logger_.debug("No bootstrapping or leaving nodes -> empty pending ranges for {}", table);
@ -726,17 +726,17 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
// get all ranges that will be affected by leaving nodes
Set<Range> affectedRanges = new HashSet<Range>();
for (InetAddress endPoint : leavingEndPoints)
affectedRanges.addAll(addressRanges.get(endPoint));
for (InetAddress endpoint : leavingEndpoints)
affectedRanges.addAll(addressRanges.get(endpoint));
// for each of those ranges, find what new nodes will be responsible for the range when
// all leaving nodes are gone.
for (Range range : affectedRanges)
{
List<InetAddress> currentEndPoints = strategy.getNaturalEndpoints(range.right, tm, table);
List<InetAddress> newEndPoints = strategy.getNaturalEndpoints(range.right, allLeftMetadata, table);
newEndPoints.removeAll(currentEndPoints);
pendingRanges.putAll(range, newEndPoints);
List<InetAddress> currentEndpoints = strategy.getNaturalEndpoints(range.right, tm, table);
List<InetAddress> newEndpoints = strategy.getNaturalEndpoints(range.right, allLeftMetadata, table);
newEndpoints.removeAll(currentEndpoints);
pendingRanges.putAll(range, newEndpoints);
}
// At this stage pendingRanges has been updated according to leave operations. We can
@ -746,12 +746,12 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
// allLeftMetadata and check in between what their ranges would be.
for (Map.Entry<Token, InetAddress> entry : bootstrapTokens.entrySet())
{
InetAddress endPoint = entry.getValue();
InetAddress endpoint = entry.getValue();
allLeftMetadata.updateNormalToken(entry.getKey(), endPoint);
for (Range range : strategy.getAddressRanges(allLeftMetadata, table).get(endPoint))
pendingRanges.put(range, endPoint);
allLeftMetadata.removeEndpoint(endPoint);
allLeftMetadata.updateNormalToken(entry.getKey(), endpoint);
for (Range range : strategy.getAddressRanges(allLeftMetadata, table).get(endpoint))
pendingRanges.put(range, endpoint);
allLeftMetadata.removeEndpoint(endpoint);
}
tm.setPendingRanges(table, pendingRanges);
@ -761,7 +761,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
}
/**
* Called when an endPoint is removed from the ring without proper
* Called when an endpoint is removed from the ring without proper
* STATE_LEAVING -> STATE_LEFT sequence. This function checks
* whether this node becomes responsible for new ranges as a
* consequence and streams data if needed.
@ -769,9 +769,9 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
* This is rather ineffective, but it does not matter so much
* since this is called very seldom
*
* @param endPoint node that has left
* @param endpoint node that has left
*/
private void restoreReplicaCount(InetAddress endPoint)
private void restoreReplicaCount(InetAddress endpoint)
{
InetAddress myAddress = FBUtilities.getLocalAddress();
@ -779,7 +779,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
{
// get all ranges that change ownership (that is, a node needs
// to take responsibility for new range)
Multimap<Range, InetAddress> changedRanges = getChangedRangesForLeaving(table, endPoint);
Multimap<Range, InetAddress> changedRanges = getChangedRangesForLeaving(table, endpoint);
// check if any of these ranges are coming our way
Set<Range> myNewRanges = new HashSet<Range>();
@ -792,7 +792,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
if (!myNewRanges.isEmpty())
{
if (logger_.isDebugEnabled())
logger_.debug(endPoint + " was removed, my added ranges: " + StringUtils.join(myNewRanges, ", "));
logger_.debug(endpoint + " was removed, my added ranges: " + StringUtils.join(myNewRanges, ", "));
Multimap<Range, InetAddress> rangeAddresses = getReplicationStrategy(table).getRangeAddresses(tokenMetadata_, table);
Multimap<InetAddress, Range> sourceRanges = HashMultimap.create();
@ -801,13 +801,13 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
// find alive sources for our new ranges
for (Range myNewRange : myNewRanges)
{
List<InetAddress> sources = DatabaseDescriptor.getEndPointSnitch().getSortedListByProximity(myAddress, rangeAddresses.get(myNewRange));
List<InetAddress> sources = DatabaseDescriptor.getEndpointSnitch().getSortedListByProximity(myAddress, rangeAddresses.get(myNewRange));
assert (!sources.contains(myAddress));
for (InetAddress source : sources)
{
if (source.equals(endPoint))
if (source.equals(endpoint))
continue;
if (failureDetector.isAlive(source))
@ -834,7 +834,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
private Multimap<Range, InetAddress> getChangedRangesForLeaving(String table, InetAddress endpoint)
{
// First get all ranges the leaving endpoint is responsible for
Collection<Range> ranges = getRangesForEndPoint(table, endpoint);
Collection<Range> ranges = getRangesForEndpoint(table, endpoint);
if (logger_.isDebugEnabled())
logger_.debug("Node " + endpoint + " ranges [" + StringUtils.join(ranges, ", ") + "]");
@ -954,24 +954,24 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
return stringify(Gossiper.instance.getUnreachableMembers());
}
private Set<String> stringify(Set<InetAddress> endPoints)
private Set<String> stringify(Set<InetAddress> endpoints)
{
Set<String> stringEndPoints = new HashSet<String>();
for (InetAddress ep : endPoints)
Set<String> stringEndpoints = new HashSet<String>();
for (InetAddress ep : endpoints)
{
stringEndPoints.add(ep.getHostAddress());
stringEndpoints.add(ep.getHostAddress());
}
return stringEndPoints;
return stringEndpoints;
}
private List<String> stringify(List<InetAddress> endPoints)
private List<String> stringify(List<InetAddress> endpoints)
{
List<String> stringEndPoints = new ArrayList<String>();
for (InetAddress ep : endPoints)
List<String> stringEndpoints = new ArrayList<String>();
for (InetAddress ep : endpoints)
{
stringEndPoints.add(ep.getHostAddress());
stringEndpoints.add(ep.getHostAddress());
}
return stringEndPoints;
return stringEndpoints;
}
public int getCurrentGenerationNumber()
@ -1104,7 +1104,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
InetAddress getPredecessor(InetAddress ep)
{
Token token = tokenMetadata_.getToken(ep);
return tokenMetadata_.getEndPoint(tokenMetadata_.getPredecessor(token));
return tokenMetadata_.getEndpoint(tokenMetadata_.getPredecessor(token));
}
/*
@ -1114,7 +1114,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
public InetAddress getSuccessor(InetAddress ep)
{
Token token = tokenMetadata_.getToken(ep);
return tokenMetadata_.getEndPoint(tokenMetadata_.getSuccessor(token));
return tokenMetadata_.getEndpoint(tokenMetadata_.getSuccessor(token));
}
/**
@ -1122,7 +1122,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
* @param ep endpoint we are interested in.
* @return range for the specified endpoint.
*/
public Range getPrimaryRangeForEndPoint(InetAddress ep)
public Range getPrimaryRangeForEndpoint(InetAddress ep)
{
return tokenMetadata_.getPrimaryRangeFor(tokenMetadata_.getToken(ep));
}
@ -1132,7 +1132,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
* @param ep endpoint we are interested in.
* @return ranges for the specified endpoint.
*/
Collection<Range> getRangesForEndPoint(String table, InetAddress ep)
Collection<Range> getRangesForEndpoint(String table, InetAddress ep)
{
return getReplicationStrategy(table).getAddressRanges(table).get(ep);
}
@ -1216,10 +1216,10 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
/**
* This function finds the closest live endpoint that contains a given key.
*/
public InetAddress findSuitableEndPoint(String table, byte[] key) throws IOException, UnavailableException
public InetAddress findSuitableEndpoint(String table, byte[] key) throws IOException, UnavailableException
{
List<InetAddress> endpoints = getNaturalEndpoints(table, key);
DatabaseDescriptor.getEndPointSnitch().sortByProximity(FBUtilities.getLocalAddress(), endpoints);
DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), endpoints);
for (InetAddress endpoint : endpoints)
{
if (FailureDetector.instance.isAlive(endpoint))
@ -1233,7 +1233,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
HashMap<String, String> map = new HashMap<String, String>();
for (Token t : tokenMetadata_.sortedTokens())
{
map.put(t.toString(), tokenMetadata_.getEndPoint(t).getHostAddress());
map.put(t.toString(), tokenMetadata_.getEndpoint(t).getHostAddress());
}
return map;
}
@ -1306,7 +1306,7 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
private void startLeaving()
{
Gossiper.instance.addLocalApplicationState(MOVE_STATE, new ApplicationState(STATE_LEAVING + Delimiter + getLocalToken().toString()));
tokenMetadata_.addLeavingEndPoint(FBUtilities.getLocalAddress());
tokenMetadata_.addLeavingEndpoint(FBUtilities.getLocalAddress());
calculatePendingRanges();
}
@ -1470,18 +1470,18 @@ public class StorageService implements IEndPointStateChangeSubscriber, StorageSe
// cannot find the endpoint for this token from metadata, but
// that would prevent this command from being issued by a node
// that has never seen the failed node.
InetAddress endPoint = tokenMetadata_.getEndPoint(token);
if (endPoint != null)
InetAddress endpoint = tokenMetadata_.getEndpoint(token);
if (endpoint != null)
{
if (endPoint.equals(FBUtilities.getLocalAddress()))
if (endpoint.equals(FBUtilities.getLocalAddress()))
throw new UnsupportedOperationException("Cannot remove node's own token");
// Let's make sure however that we're not removing a live
// token (member)
if (Gossiper.instance.getLiveMembers().contains(endPoint))
throw new UnsupportedOperationException("Node " + endPoint + " is alive and owns this token. Use decommission command to remove it from the ring");
if (Gossiper.instance.getLiveMembers().contains(endpoint))
throw new UnsupportedOperationException("Node " + endpoint + " is alive and owns this token. Use decommission command to remove it from the ring");
removeEndPointLocally(endPoint);
removeEndpointLocally(endpoint);
calculatePendingRanges();
}

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

@ -62,7 +62,7 @@ public interface StorageServiceMBean
*
* @return mapping of ranges to end points
*/
public Map<Range, List<String>> getRangeToEndPointMap(String keyspace);
public Map<Range, List<String>> getRangeToEndpointMap(String keyspace);
/**
* Numeric load value.

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

@ -607,7 +607,7 @@ public class CassandraServer implements Cassandra.Iface
public List<TokenRange> describe_ring(String keyspace)
{
List<TokenRange> ranges = new ArrayList<TokenRange>();
for (Map.Entry<Range, List<String>> entry : StorageService.instance.getRangeToEndPointMap(keyspace).entrySet())
for (Map.Entry<Range, List<String>> entry : StorageService.instance.getRangeToEndpointMap(keyspace).entrySet())
{
Range range = entry.getKey();
List<String> endpoints = entry.getValue();

6
src/java/org/apache/cassandra/tools/ClusterCmd.java
View File

@ -144,9 +144,9 @@ public class ClusterCmd {
hf.printHelp(usage, "", options, header);
}
public void printEndPoints(String keyspace, String key)
public void printEndpoints(String keyspace, String key)
{
List<InetAddress> endpoints = probe.getEndPoints(keyspace, key);
List<InetAddress> endpoints = probe.getEndpoints(keyspace, key);
System.out.println(String.format("%-17s: %s", "Key", key));
System.out.println(String.format("%-17s: %s", "Endpoints", endpoints));
}
@ -260,7 +260,7 @@ public class ClusterCmd {
ClusterCmd.printUsage();
System.exit(1);
}
clusterCmd.printEndPoints(arguments[1], arguments[2]);
clusterCmd.printEndpoints(arguments[1], arguments[2]);
}
else if (cmdName.equals("global_snapshot"))
{

2
src/java/org/apache/cassandra/tools/NodeCmd.java
View File

@ -92,7 +92,7 @@ public class NodeCmd {
*/
public void printRing(PrintStream outs)
{
Map<Range, List<String>> rangeMap = probe.getRangeToEndPointMap(null);
Map<Range, List<String>> rangeMap = probe.getRangeToEndpointMap(null);
List<Range> ranges = new ArrayList<Range>(rangeMap.keySet());
Collections.sort(ranges);
Set<String> liveNodes = probe.getLiveNodes();

8
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -152,9 +152,9 @@ public class NodeProbe
ssProxy.drain();
}
public Map<Range, List<String>> getRangeToEndPointMap(String tableName)
public Map<Range, List<String>> getRangeToEndpointMap(String tableName)
{
return ssProxy.getRangeToEndPointMap(tableName);
return ssProxy.getRangeToEndpointMap(tableName);
}
public Set<String> getLiveNodes()
@ -169,7 +169,7 @@ public class NodeProbe
*/
public void printRing(PrintStream outs)
{
Map<Range, List<String>> rangeMap = ssProxy.getRangeToEndPointMap(null);
Map<Range, List<String>> rangeMap = ssProxy.getRangeToEndpointMap(null);
List<Range> ranges = new ArrayList<Range>(rangeMap.keySet());
Collections.sort(ranges);
Set<String> liveNodes = ssProxy.getLiveNodes();
@ -407,7 +407,7 @@ public class NodeProbe
}
}
public List<InetAddress> getEndPoints(String keyspace, String key)
public List<InetAddress> getEndpoints(String keyspace, String key)
{
// FIXME: string key
return ssProxy.getNaturalEndpoints(keyspace, key.getBytes(UTF8));

8
test/conf/cassandra.xml
View File

@ -37,7 +37,7 @@
<DiskAccessMode>mmap</DiskAccessMode>
<MemtableThroughputInMB>1</MemtableThroughputInMB>
<MemtableOperationsInMillions>0.00002</MemtableOperationsInMillions> <!-- 20 -->
<EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
<EndpointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndpointSnitch>
<Keyspaces>
<Keyspace Name = "Keyspace1">
<ColumnFamily Name="Standard1" RowsCached="10%" KeysCached="0"/>
@ -58,13 +58,13 @@
<ColumnFamily ColumnType="Super" CompareSubcolumnsWith="TimeUUIDType" Name="Super4"/>
<ReplicaPlacementStrategy>org.apache.cassandra.locator.RackUnawareStrategy</ReplicaPlacementStrategy>
<ReplicationFactor>1</ReplicationFactor>
<EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
<EndpointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndpointSnitch>
</Keyspace>
<Keyspace Name = "Keyspace3">
<ColumnFamily Name="Standard1"/>
<ReplicaPlacementStrategy>org.apache.cassandra.locator.RackUnawareStrategy</ReplicaPlacementStrategy>
<ReplicationFactor>5</ReplicationFactor>
<EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
<EndpointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndpointSnitch>
</Keyspace>
<Keyspace Name = "Keyspace4">
<ColumnFamily Name="Standard1"/>
@ -73,7 +73,7 @@
<ColumnFamily ColumnType="Super" CompareSubcolumnsWith="TimeUUIDType" Name="Super4"/>
<ReplicaPlacementStrategy>org.apache.cassandra.locator.RackUnawareStrategy</ReplicaPlacementStrategy>
<ReplicationFactor>3</ReplicationFactor>
<EndPointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndPointSnitch>
<EndpointSnitch>org.apache.cassandra.locator.EndPointSnitch</EndpointSnitch>
</Keyspace>
</Keyspaces>
<Seeds>

2
test/system/__init__.py
View File

@ -142,7 +142,7 @@ class ThriftTester(BaseTester):
self.client.transport.close()
def define_schema(self):
keyspace1 = Cassandra.KsDef('Keyspace1', 'org.apache.cassandra.locator.RackUnawareStrategy', 1, 'org.apache.cassandra.locator.EndPointSnitch',
keyspace1 = Cassandra.KsDef('Keyspace1', 'org.apache.cassandra.locator.RackUnawareStrategy', 1, 'org.apache.cassandra.locator.EndPointSnitch',
[
Cassandra.CfDef('Keyspace1', 'Standard1'),
Cassandra.CfDef('Keyspace1', 'Standard2'),

10
test/unit/org/apache/cassandra/client/TestRingCache.java
View File

@ -92,14 +92,14 @@ public class TestRingCache
ColumnPath col = new ColumnPath("Standard1").setSuper_column(null).setColumn("col1".getBytes());
ColumnParent parent = new ColumnParent("Standard1").setSuper_column(null);
List<InetAddress> endPoints = tester.ringCache.getEndPoint(row);
List<InetAddress> endpoints = tester.ringCache.getEndpoint(row);
String hosts="";
for (int i = 0; i < endPoints.size(); i++)
hosts = hosts + ((i > 0) ? "," : "") + endPoints.get(i);
System.out.println("hosts with key " + new String(row) + " : " + hosts + "; choose " + endPoints.get(0));
for (int i = 0; i < endpoints.size(); i++)
hosts = hosts + ((i > 0) ? "," : "") + endpoints.get(i);
System.out.println("hosts with key " + new String(row) + " : " + hosts + "; choose " + endpoints.get(0));
// now, read the row back directly from the host owning the row locally
tester.setup(endPoints.get(0).getHostAddress(), DatabaseDescriptor.getRpcPort());
tester.setup(endpoints.get(0).getHostAddress(), DatabaseDescriptor.getRpcPort());
tester.thriftClient.insert(keyspace, row, parent, new Column("col1".getBytes(), "val1".getBytes(), 1), ConsistencyLevel.ONE);
Column column = tester.thriftClient.get(keyspace, row, col, ConsistencyLevel.ONE).column;
System.out.println("read row " + new String(row) + " " + new String(column.name) + ":" + new String(column.value) + ":" + column.timestamp);

2
test/unit/org/apache/cassandra/dht/BootStrapperTest.java
View File

@ -59,7 +59,7 @@ public class BootStrapperTest
assert three.equals(source);
InetAddress myEndpoint = InetAddress.getByName("127.0.0.1");
Range range3 = ss.getPrimaryRangeForEndPoint(three);
Range range3 = ss.getPrimaryRangeForEndpoint(three);
Token fakeToken = ((IPartitioner)StorageService.getPartitioner()).midpoint(range3.left, range3.right);
assert range3.contains(fakeToken);
ss.onChange(myEndpoint, StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_BOOTSTRAPPING + StorageService.Delimiter + ss.getPartitioner().getTokenFactory().toString(fakeToken)));

6
test/unit/org/apache/cassandra/gms/GossipDigestTest.java
View File

@ -37,12 +37,12 @@ public class GossipDigestTest
@Test
public void test() throws IOException
{
InetAddress endPoint = InetAddress.getByName("127.0.0.1");
InetAddress endpoint = InetAddress.getByName("127.0.0.1");
int generation = 0;
int maxVersion = 123;
GossipDigest expected = new GossipDigest(endPoint, generation, maxVersion);
GossipDigest expected = new GossipDigest(endpoint, generation, maxVersion);
//make sure we get the same values out
assertEquals(endPoint, expected.getEndPoint());
assertEquals(endpoint, expected.getEndpoint());
assertEquals(generation, expected.getGeneration());
assertEquals(maxVersion, expected.getMaxVersion());

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

@ -36,7 +36,7 @@ import org.junit.Test;
public class RackAwareStrategyTest
{
private List<Token> endPointTokens;
private List<Token> endpointTokens;
private List<Token> keyTokens;
private TokenMetadata tmd;
private Map<String, ArrayList<InetAddress>> expectedResults;
@ -44,7 +44,7 @@ public class RackAwareStrategyTest
@Before
public void init()
{
endPointTokens = new ArrayList<Token>();
endpointTokens = new ArrayList<Token>();
keyTokens = new ArrayList<Token>();
tmd = new TokenMetadata();
expectedResults = new HashMap<String, ArrayList<InetAddress>>();
@ -58,13 +58,13 @@ public class RackAwareStrategyTest
@Test
public void testBigIntegerEndpointsA() throws UnknownHostException
{
EndPointSnitch endPointSnitch = new EndPointSnitch();
EndPointSnitch endpointSnitch = new EndPointSnitch();
AbstractReplicationStrategy strategy = new RackAwareStrategy(tmd, endPointSnitch);
addEndPoint("0", "5", "254.0.0.1");
addEndPoint("10", "15", "254.0.0.2");
addEndPoint("20", "25", "254.0.0.3");
addEndPoint("30", "35", "254.0.0.4");
AbstractReplicationStrategy strategy = new RackAwareStrategy(tmd, endpointSnitch);
addEndpoint("0", "5", "254.0.0.1");
addEndpoint("10", "15", "254.0.0.2");
addEndpoint("20", "25", "254.0.0.3");
addEndpoint("30", "35", "254.0.0.4");
expectedResults.put("5", buildResult("254.0.0.2", "254.0.0.3", "254.0.0.4"));
expectedResults.put("15", buildResult("254.0.0.3", "254.0.0.4", "254.0.0.1"));
@ -83,13 +83,13 @@ public class RackAwareStrategyTest
@Test
public void testBigIntegerEndpointsB() throws UnknownHostException
{
EndPointSnitch endPointSnitch = new EndPointSnitch();
EndPointSnitch endpointSnitch = new EndPointSnitch();
AbstractReplicationStrategy strategy = new RackAwareStrategy(tmd, endPointSnitch);
addEndPoint("0", "5", "254.0.0.1");
addEndPoint("10", "15", "254.0.0.2");
addEndPoint("20", "25", "254.1.0.3");
addEndPoint("30", "35", "254.0.0.4");
AbstractReplicationStrategy strategy = new RackAwareStrategy(tmd, endpointSnitch);
addEndpoint("0", "5", "254.0.0.1");
addEndpoint("10", "15", "254.0.0.2");
addEndpoint("20", "25", "254.1.0.3");
addEndpoint("30", "35", "254.0.0.4");
expectedResults.put("5", buildResult("254.0.0.2", "254.1.0.3", "254.0.0.4"));
expectedResults.put("15", buildResult("254.1.0.3", "254.0.0.4", "254.0.0.1"));
@ -109,13 +109,13 @@ public class RackAwareStrategyTest
@Test
public void testBigIntegerEndpointsC() throws UnknownHostException
{
EndPointSnitch endPointSnitch = new EndPointSnitch();
EndPointSnitch endpointSnitch = new EndPointSnitch();
AbstractReplicationStrategy strategy = new RackAwareStrategy(tmd, endPointSnitch);
addEndPoint("0", "5", "254.0.0.1");
addEndPoint("10", "15", "254.0.0.2");
addEndPoint("20", "25", "254.0.1.3");
addEndPoint("30", "35", "254.1.0.4");
AbstractReplicationStrategy strategy = new RackAwareStrategy(tmd, endpointSnitch);
addEndpoint("0", "5", "254.0.0.1");
addEndpoint("10", "15", "254.0.0.2");
addEndpoint("20", "25", "254.0.1.3");
addEndpoint("30", "35", "254.1.0.4");
expectedResults.put("5", buildResult("254.0.0.2", "254.0.1.3", "254.1.0.4"));
expectedResults.put("15", buildResult("254.0.1.3", "254.1.0.4", "254.0.0.1"));
@ -144,27 +144,27 @@ public class RackAwareStrategyTest
return result;
}
private void addEndPoint(String endPointTokenID, String keyTokenID, String endPointAddress) throws UnknownHostException
private void addEndpoint(String endpointTokenID, String keyTokenID, String endpointAddress) throws UnknownHostException
{
BigIntegerToken endPointToken = new BigIntegerToken(endPointTokenID);
endPointTokens.add(endPointToken);
BigIntegerToken endpointToken = new BigIntegerToken(endpointTokenID);
endpointTokens.add(endpointToken);
BigIntegerToken keyToken = new BigIntegerToken(keyTokenID);
keyTokens.add(keyToken);
InetAddress ep = InetAddress.getByName(endPointAddress);
tmd.updateNormalToken(endPointToken, ep);
InetAddress ep = InetAddress.getByName(endpointAddress);
tmd.updateNormalToken(endpointToken, ep);
}
private void testGetEndpoints(AbstractReplicationStrategy strategy, Token[] keyTokens, String table) throws UnknownHostException
{
for (Token keyToken : keyTokens)
{
List<InetAddress> endPoints = strategy.getNaturalEndpoints(keyToken, tmd, table);
for (int j = 0; j < endPoints.size(); j++)
List<InetAddress> endpoints = strategy.getNaturalEndpoints(keyToken, tmd, table);
for (int j = 0; j < endpoints.size(); j++)
{
ArrayList<InetAddress> hostsExpected = expectedResults.get(keyToken.toString());
assertEquals(endPoints.get(j), hostsExpected.get(j));
assertEquals(endpoints.get(j), hostsExpected.get(j));
}
}
}

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

@ -65,14 +65,14 @@ public class RackUnawareStrategyTest extends SchemaLoader
TokenMetadata tmd = new TokenMetadata();
AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, null);
List<Token> endPointTokens = new ArrayList<Token>();
List<Token> endpointTokens = new ArrayList<Token>();
List<Token> keyTokens = new ArrayList<Token>();
for (int i = 0; i < 5; i++) {
endPointTokens.add(new BigIntegerToken(String.valueOf(10 * i)));
endpointTokens.add(new BigIntegerToken(String.valueOf(10 * i)));
keyTokens.add(new BigIntegerToken(String.valueOf(10 * i + 5)));
}
for (String table : DatabaseDescriptor.getNonSystemTables())
testGetEndpoints(tmd, strategy, endPointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0]), table);
testGetEndpoints(tmd, strategy, endpointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0]), table);
}
@Test
@ -82,35 +82,35 @@ public class RackUnawareStrategyTest extends SchemaLoader
IPartitioner partitioner = new OrderPreservingPartitioner();
AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, null);
List<Token> endPointTokens = new ArrayList<Token>();
List<Token> endpointTokens = new ArrayList<Token>();
List<Token> keyTokens = new ArrayList<Token>();
for (int i = 0; i < 5; i++) {
endPointTokens.add(new StringToken(String.valueOf((char)('a' + i * 2))));
endpointTokens.add(new StringToken(String.valueOf((char)('a' + i * 2))));
keyTokens.add(partitioner.getToken(String.valueOf((char)('a' + i * 2 + 1)).getBytes()));
}
for (String table : DatabaseDescriptor.getNonSystemTables())
testGetEndpoints(tmd, strategy, endPointTokens.toArray(new Token[0]), keyTokens.toArray(new Token[0]), table);
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, String table) throws UnknownHostException
private void testGetEndpoints(TokenMetadata tmd, AbstractReplicationStrategy strategy, Token[] endpointTokens, Token[] keyTokens, String table) throws UnknownHostException
{
List<InetAddress> hosts = new ArrayList<InetAddress>();
for (int i = 0; i < endPointTokens.length; i++)
for (int i = 0; i < endpointTokens.length; i++)
{
InetAddress ep = InetAddress.getByName("127.0.0." + String.valueOf(i + 1));
tmd.updateNormalToken(endPointTokens[i], ep);
tmd.updateNormalToken(endpointTokens[i], ep);
hosts.add(ep);
}
for (int i = 0; i < keyTokens.length; i++)
{
List<InetAddress> endPoints = strategy.getNaturalEndpoints(keyTokens[i], table);
assertEquals(DatabaseDescriptor.getReplicationFactor(table), endPoints.size());
for (int j = 0; j < endPoints.size(); j++)
List<InetAddress> 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()));
assertEquals(endpoints.get(j), hosts.get((i + j + 1) % hosts.size()));
}
}
}
@ -124,27 +124,27 @@ public class RackUnawareStrategyTest extends SchemaLoader
TokenMetadata oldTmd = StorageServiceAccessor.setTokenMetadata(tmd);
AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, null);
Token[] endPointTokens = new Token[RING_SIZE];
Token[] endpointTokens = new Token[RING_SIZE];
Token[] keyTokens = new Token[RING_SIZE];
for (int i = 0; i < RING_SIZE; i++)
{
endPointTokens[i] = new BigIntegerToken(String.valueOf(RING_SIZE * 2 * i));
endpointTokens[i] = new BigIntegerToken(String.valueOf(RING_SIZE * 2 * i));
keyTokens[i] = new BigIntegerToken(String.valueOf(RING_SIZE * 2 * i + RING_SIZE));
}
List<InetAddress> hosts = new ArrayList<InetAddress>();
for (int i = 0; i < endPointTokens.length; i++)
for (int i = 0; i < endpointTokens.length; i++)
{
InetAddress ep = InetAddress.getByName("127.0.0." + String.valueOf(i + 1));
tmd.updateNormalToken(endPointTokens[i], ep);
tmd.updateNormalToken(endpointTokens[i], ep);
hosts.add(ep);
}
// 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);
InetAddress bootstrapEndpoint = InetAddress.getByName("127.0.0.11");
tmd.addBootstrapToken(bsToken, bootstrapEndpoint);
for (String table : DatabaseDescriptor.getNonSystemTables())
{
@ -153,20 +153,20 @@ public class RackUnawareStrategyTest extends SchemaLoader
for (int i = 0; i < keyTokens.length; i++)
{
Collection<InetAddress> endPoints = strategy.getWriteEndpoints(keyTokens[i], table, strategy.getNaturalEndpoints(keyTokens[i], table));
assertTrue(endPoints.size() >= replicationFactor);
Collection<InetAddress> 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())));
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.
// 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));
assertFalse(endpoints.contains(bootstrapEndpoint));
else
assertTrue(endPoints.contains(bootstrapEndPoint));
assertTrue(endpoints.contains(bootstrapEndpoint));
}
}

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

@ -60,7 +60,7 @@ public class MoveTest
* StorageService.onChange and does not manipulate token metadata directly.
*/
@Test
public void testWriteEndPointsDuringLeave() throws UnknownHostException
public void testWriteEndpointsDuringLeave() throws UnknownHostException
{
StorageService ss = StorageService.instance;
final int RING_SIZE = 5;
@ -74,11 +74,11 @@ public class MoveTest
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategies = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endPointTokens = new ArrayList<Token>();
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
List<InetAddress> hosts = new ArrayList<InetAddress>();
createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, RING_SIZE);
createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, RING_SIZE);
final Map<String, List<Range>> deadNodesRanges = new HashMap<String, List<Range>>();
for (String table : DatabaseDescriptor.getNonSystemTables())
@ -90,7 +90,7 @@ public class MoveTest
}
// Third node leaves
ss.onChange(hosts.get(LEAVING_NODE), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(LEAVING_NODE))));
ss.onChange(hosts.get(LEAVING_NODE), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endpointTokens.get(LEAVING_NODE))));
// check that it is correctly marked as leaving in tmd
assertTrue(tmd.isLeaving(hosts.get(LEAVING_NODE)));
@ -117,14 +117,14 @@ public class MoveTest
final int replicaStart = (LEAVING_NODE-replicationFactor+RING_SIZE)%RING_SIZE;
for (int i=0; i<keyTokens.size(); ++i)
{
Collection<InetAddress> endPoints = testStrategy.getWriteEndpoints(keyTokens.get(i), table, testStrategy.getNaturalEndpoints(keyTokens.get(i), table));
Collection<InetAddress> 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);
assertTrue(endpoints.size() == replicationFactor + 1);
else
assertTrue(endPoints.size() == replicationFactor);
assertTrue(endpoints.size() == replicationFactor);
}
}
@ -150,17 +150,17 @@ public class MoveTest
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endPointTokens = new ArrayList<Token>();
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
List<InetAddress> hosts = new ArrayList<InetAddress>();
// create a ring or 10 nodes
createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, RING_SIZE);
createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, RING_SIZE);
// nodes 6, 8 and 9 leave
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))));
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");
@ -168,7 +168,7 @@ public class MoveTest
InetAddress boot2 = InetAddress.getByName("127.0.1.2");
ss.onChange(boot2, StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_BOOTSTRAPPING + StorageService.Delimiter + partitioner.getTokenFactory().toString(keyTokens.get(7))));
Collection<InetAddress> endPoints = null;
Collection<InetAddress> endpoints = null;
// pre-calculate the results.
Map<String, Multimap<Token, InetAddress>> expectedEndpoints = new HashMap<String, Multimap<Token, InetAddress>>();
@ -221,9 +221,9 @@ public class MoveTest
{
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));
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:
@ -232,84 +232,84 @@ public class MoveTest
// 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)));
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));
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));
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));
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));
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));
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)));
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(0)));
assertTrue(endPoints.contains(hosts.get(1)));
assertTrue(endPoints.contains(hosts.get(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)));
}
// 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(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(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))));
// adjust precalcuated results. this changes what the epected endpoints are.
@ -336,9 +336,9 @@ public class MoveTest
{
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));
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)
@ -347,67 +347,67 @@ public class MoveTest
// 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)));
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(4)));
assertTrue(endPoints.contains(hosts.get(5)));
assertTrue(endPoints.contains(boot1));
endpoints = testStrategy.getWriteEndpoints(keyTokens.get(3), table, testStrategy.getNaturalEndpoints(keyTokens.get(3), table));
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), 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)));
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)));
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)));
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)));
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)));
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)));
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)));
}
ss.setPartitionerUnsafe(oldPartitioner);
@ -426,15 +426,15 @@ public class MoveTest
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endPointTokens = new ArrayList<Token>();
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
List<InetAddress> hosts = new ArrayList<InetAddress>();
// create a ring or 5 nodes
createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, 5);
createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, 5);
// node 2 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(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endpointTokens.get(2))));
// don't bother to test pending ranges here, that is extensively tested by other
// tests. Just check that the node is in appropriate lists.
@ -495,23 +495,23 @@ public class MoveTest
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endPointTokens = new ArrayList<Token>();
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
List<InetAddress> hosts = new ArrayList<InetAddress>();
// create a ring or 5 nodes
createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, 5);
createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, 5);
// node 2 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(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(endpointTokens.get(2))));
assertTrue(tmd.isLeaving(hosts.get(2)));
assertTrue(tmd.getToken(hosts.get(2)).equals(endPointTokens.get(2)));
assertTrue(tmd.getToken(hosts.get(2)).equals(endpointTokens.get(2)));
// back to normal
ss.onChange(hosts.get(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_NORMAL + StorageService.Delimiter + partitioner.getTokenFactory().toString(keyTokens.get(2))));
assertTrue(tmd.getLeavingEndPoints().isEmpty());
assertTrue(tmd.getLeavingEndpoints().isEmpty());
assertTrue(tmd.getToken(hosts.get(2)).equals(keyTokens.get(2)));
// node 3 goes through leave and left and then jumps to normal
@ -520,7 +520,7 @@ public class MoveTest
ss.onChange(hosts.get(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_NORMAL + StorageService.Delimiter + partitioner.getTokenFactory().toString(keyTokens.get(4))));
assertTrue(tmd.getBootstrapTokens().isEmpty());
assertTrue(tmd.getLeavingEndPoints().isEmpty());
assertTrue(tmd.getLeavingEndpoints().isEmpty());
assertTrue(tmd.getToken(hosts.get(2)).equals(keyTokens.get(4)));
ss.setPartitionerUnsafe(oldPartitioner);
@ -539,19 +539,19 @@ public class MoveTest
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endPointTokens = new ArrayList<Token>();
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
List<InetAddress> hosts = new ArrayList<InetAddress>();
// create a ring or 5 nodes
createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, 5);
createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, 5);
// node 2 leaves with _different_ token
ss.onChange(hosts.get(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(keyTokens.get(0))));
assertTrue(tmd.getToken(hosts.get(2)).equals(keyTokens.get(0)));
assertTrue(tmd.isLeaving(hosts.get(2)));
assertTrue(tmd.getEndPoint(endPointTokens.get(2)) == null);
assertTrue(tmd.getEndpoint(endpointTokens.get(2)) == null);
// go to boostrap
ss.onChange(hosts.get(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_BOOTSTRAPPING + StorageService.Delimiter + partitioner.getTokenFactory().toString(keyTokens.get(1))));
@ -563,7 +563,7 @@ public class MoveTest
// jump to leaving again
ss.onChange(hosts.get(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEAVING + StorageService.Delimiter + partitioner.getTokenFactory().toString(keyTokens.get(1))));
assertTrue(tmd.getEndPoint(keyTokens.get(1)).equals(hosts.get(2)));
assertTrue(tmd.getEndpoint(keyTokens.get(1)).equals(hosts.get(2)));
assertTrue(tmd.isLeaving(hosts.get(2)));
assertTrue(tmd.getBootstrapTokens().isEmpty());
@ -589,15 +589,15 @@ public class MoveTest
IPartitioner oldPartitioner = ss.setPartitionerUnsafe(partitioner);
Map<String, AbstractReplicationStrategy> oldStrategy = ss.setReplicationStrategyUnsafe(createReplacements(testStrategy));
ArrayList<Token> endPointTokens = new ArrayList<Token>();
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
List<InetAddress> hosts = new ArrayList<InetAddress>();
// create a ring or 5 nodes
createInitialRing(ss, partitioner, endPointTokens, keyTokens, hosts, 5);
createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, 5);
// node hosts.get(2) goes jumps to left
ss.onChange(hosts.get(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEFT + StorageService.Delimiter + StorageService.LEFT_NORMALLY + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(2))));
ss.onChange(hosts.get(2), StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_LEFT + StorageService.Delimiter + StorageService.LEFT_NORMALLY + StorageService.Delimiter + partitioner.getTokenFactory().toString(endpointTokens.get(2))));
assertFalse(tmd.isMember(hosts.get(2)));
@ -622,25 +622,25 @@ public class MoveTest
/**
* Creates initial set of nodes and tokens. Nodes are added to StorageService as 'normal'
*/
private void createInitialRing(StorageService ss, IPartitioner partitioner, List<Token> endPointTokens,
private void createInitialRing(StorageService ss, IPartitioner partitioner, List<Token> endpointTokens,
List<Token> keyTokens, List<InetAddress> hosts, int howMany)
throws UnknownHostException
{
for (int i=0; i<howMany; i++)
{
endPointTokens.add(new BigIntegerToken(String.valueOf(10 * i)));
endpointTokens.add(new BigIntegerToken(String.valueOf(10 * i)));
keyTokens.add(new BigIntegerToken(String.valueOf(10 * i + 5)));
}
for (int i=0; i<endPointTokens.size(); i++)
for (int i=0; i<endpointTokens.size(); i++)
{
InetAddress ep = InetAddress.getByName("127.0.0." + String.valueOf(i + 1));
ss.onChange(ep, StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_NORMAL + StorageService.Delimiter + partitioner.getTokenFactory().toString(endPointTokens.get(i))));
ss.onChange(ep, StorageService.MOVE_STATE, new ApplicationState(StorageService.STATE_NORMAL + StorageService.Delimiter + partitioner.getTokenFactory().toString(endpointTokens.get(i))));
hosts.add(ep);
}
// check that all nodes are in token metadata
for (int i=0; i<endPointTokens.size(); ++i)
for (int i=0; i<endpointTokens.size(); ++i)
assertTrue(ss.getTokenMetadata().isMember(hosts.get(i)));
}