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

Merge branch 'cassandra-2.2' into cassandra-3.0

This commit is contained in:
Aleksey Yeschenko 2015-09-16 12:04:53 +01:00
parent c694cd62af 5b05b68264
commit 8134f09c2b
6 changed files with 335 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -55,6 +55,7 @@ Merged from 2.1:
* (cqlsh) update list of CQL keywords (CASSANDRA-9232)
* Add nodetool gettraceprobability command (CASSANDRA-10234)
Merged from 2.0:
* Fix consolidating racks violating the RF contract (CASSANDRA-10238)
* Disallow decommission when node is in drained state (CASSANDRA-8741)

1
src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java
View File

@ -83,6 +83,7 @@ public class GossipingPropertyFileSnitch extends AbstractNetworkTopologySnitch//
protected void runMayThrow() throws ConfigurationException
{
reloadConfiguration();
StorageService.instance.updateTopology(FBUtilities.getBroadcastAddress());
}
};
ResourceWatcher.watch(SnitchProperties.RACKDC_PROPERTY_FILENAME, runnable, refreshPeriodInSeconds * 1000);

1
src/java/org/apache/cassandra/locator/PropertyFileSnitch.java
View File

@ -70,6 +70,7 @@ public class PropertyFileSnitch extends AbstractNetworkTopologySnitch
protected void runMayThrow() throws ConfigurationException
{
reloadConfiguration();
StorageService.instance.updateTopology();
}
};
ResourceWatcher.watch(SNITCH_PROPERTIES_FILENAME, runnable, 60 * 1000);

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

@ -80,14 +80,14 @@ public class TokenMetadata
// Finally, note that recording the tokens of joining nodes in bootstrapTokens also
// means we can detect and reject the addition of multiple nodes at the same token
// before one becomes part of the ring.
private final BiMultiValMap<Token, InetAddress> bootstrapTokens = new BiMultiValMap<Token, InetAddress>();
private final BiMultiValMap<Token, InetAddress> bootstrapTokens = new BiMultiValMap<>();
// (don't need to record Token here since it's still part of tokenToEndpointMap until it's done leaving)
private final Set<InetAddress> leavingEndpoints = new HashSet<InetAddress>();
private final Set<InetAddress> leavingEndpoints = new HashSet<>();
// this is a cache of the calculation from {tokenToEndpointMap, bootstrapTokens, leavingEndpoints}
private final ConcurrentMap<String, Multimap<Range<Token>, InetAddress>> pendingRanges = new ConcurrentHashMap<String, Multimap<Range<Token>, InetAddress>>();
private final ConcurrentMap<String, Multimap<Range<Token>, InetAddress>> pendingRanges = new ConcurrentHashMap<>();
// nodes which are migrating to the new tokens in the ring
private final Set<Pair<Token, InetAddress>> movingEndpoints = new HashSet<Pair<Token, InetAddress>>();
private final Set<Pair<Token, InetAddress>> movingEndpoints = new HashSet<>();
/* Use this lock for manipulating the token map */
private final ReadWriteLock lock = new ReentrantReadWriteLock(true);
@ -135,7 +135,7 @@ public class TokenMetadata
private ArrayList<Token> sortTokens()
{
return new ArrayList<Token>(tokenToEndpointMap.keySet());
return new ArrayList<>(tokenToEndpointMap.keySet());
}
/** @return the number of nodes bootstrapping into source's primary range */
@ -179,8 +179,6 @@ public class TokenMetadata
*
* Prefer this whenever there are multiple pairs to update, as each update (whether a single or multiple)
* is expensive (CASSANDRA-3831).
*
* @param endpointTokens
*/
public void updateNormalTokens(Multimap<InetAddress, Token> endpointTokens)
{
@ -227,9 +225,6 @@ public class TokenMetadata
/**
* Store an end-point to host ID mapping. Each ID must be unique, and
* cannot be changed after the fact.
*
* @param hostId
* @param endpoint
*/
public void updateHostId(UUID hostId, InetAddress endpoint)
{
@ -298,7 +293,7 @@ public class TokenMetadata
lock.readLock().lock();
try
{
Map<InetAddress, UUID> readMap = new HashMap<InetAddress, UUID>();
Map<InetAddress, UUID> readMap = new HashMap<>();
readMap.putAll(endpointToHostIdMap);
return readMap;
}
@ -420,6 +415,43 @@ public class TokenMetadata
}
}
/**
* This is called when the snitch properties for this endpoint are updated, see CASSANDRA-10238.
*/
public void updateTopology(InetAddress endpoint)
{
assert endpoint != null;
lock.writeLock().lock();
try
{
logger.info("Updating topology for {}", endpoint);
topology.updateEndpoint(endpoint);
}
finally
{
lock.writeLock().unlock();
}
}
/**
* This is called when the snitch properties for many endpoints are updated, it will update
* the topology mappings of any endpoints whose snitch has changed, see CASSANDRA-10238.
*/
public void updateTopology()
{
lock.writeLock().lock();
try
{
logger.info("Updating topology for all endpoints that have changed");
topology.updateEndpoints();
}
finally
{
lock.writeLock().unlock();
}
}
/**
* Remove pair of token/address from moving endpoints
* @param endpoint address of the moving node
@ -456,7 +488,7 @@ public class TokenMetadata
lock.readLock().lock();
try
{
return new ArrayList<Token>(tokenToEndpointMap.inverse().get(endpoint));
return new ArrayList<>(tokenToEndpointMap.inverse().get(endpoint));
}
finally
{
@ -522,7 +554,7 @@ public class TokenMetadata
}
}
private final AtomicReference<TokenMetadata> cachedTokenMap = new AtomicReference<TokenMetadata>();
private final AtomicReference<TokenMetadata> cachedTokenMap = new AtomicReference<>();
/**
* Create a copy of TokenMetadata with only tokenToEndpointMap. That is, pending ranges,
@ -533,7 +565,7 @@ public class TokenMetadata
lock.readLock().lock();
try
{
return new TokenMetadata(SortedBiMultiValMap.<Token, InetAddress>create(tokenToEndpointMap, null, inetaddressCmp),
return new TokenMetadata(SortedBiMultiValMap.create(tokenToEndpointMap, null, inetaddressCmp),
HashBiMap.create(endpointToHostIdMap),
new Topology(topology),
partitioner);
@ -637,9 +669,9 @@ public class TokenMetadata
public Collection<Range<Token>> getPrimaryRangesFor(Collection<Token> tokens)
{
Collection<Range<Token>> ranges = new ArrayList<Range<Token>>(tokens.size());
Collection<Range<Token>> ranges = new ArrayList<>(tokens.size());
for (Token right : tokens)
ranges.add(new Range<Token>(getPredecessor(right), right));
ranges.add(new Range<>(getPredecessor(right), right));
return ranges;
}
@ -675,7 +707,7 @@ public class TokenMetadata
public List<Range<Token>> getPendingRanges(String keyspaceName, InetAddress endpoint)
{
List<Range<Token>> ranges = new ArrayList<Range<Token>>();
List<Range<Token>> ranges = new ArrayList<>();
for (Map.Entry<Range<Token>, InetAddress> entry : getPendingRangesMM(keyspaceName).entries())
{
if (entry.getValue().equals(endpoint))
@ -965,7 +997,7 @@ public class TokenMetadata
for (InetAddress ep : eps)
{
sb.append(ep);
sb.append(":");
sb.append(':');
sb.append(tokenToEndpointMap.inverse().get(ep));
sb.append(System.getProperty("line.separator"));
}
@ -977,7 +1009,7 @@ public class TokenMetadata
sb.append(System.getProperty("line.separator"));
for (Map.Entry<Token, InetAddress> entry : bootstrapTokens.entrySet())
{
sb.append(entry.getValue()).append(":").append(entry.getKey());
sb.append(entry.getValue()).append(':').append(entry.getKey());
sb.append(System.getProperty("line.separator"));
}
}
@ -1016,7 +1048,7 @@ public class TokenMetadata
{
for (Map.Entry<Range<Token>, InetAddress> rmap : entry.getValue().entries())
{
sb.append(rmap.getValue()).append(":").append(rmap.getKey());
sb.append(rmap.getValue()).append(':').append(rmap.getKey());
sb.append(System.getProperty("line.separator"));
}
}
@ -1030,7 +1062,7 @@ public class TokenMetadata
if (ranges.isEmpty())
return Collections.emptyList();
Set<InetAddress> endpoints = new HashSet<InetAddress>();
Set<InetAddress> endpoints = new HashSet<>();
for (Map.Entry<Range<Token>, Collection<InetAddress>> entry : ranges.entrySet())
{
if (entry.getKey().contains(token))
@ -1074,7 +1106,7 @@ public class TokenMetadata
lock.readLock().lock();
try
{
Map<Token, InetAddress> map = new HashMap<Token, InetAddress>(tokenToEndpointMap.size() + bootstrapTokens.size());
Map<Token, InetAddress> map = new HashMap<>(tokenToEndpointMap.size() + bootstrapTokens.size());
map.putAll(tokenToEndpointMap);
map.putAll(bootstrapTokens);
return map;
@ -1126,14 +1158,14 @@ public class TokenMetadata
/** reverse-lookup map for endpoint to current known dc/rack assignment */
private final Map<InetAddress, Pair<String, String>> currentLocations;
protected Topology()
Topology()
{
dcEndpoints = HashMultimap.create();
dcRacks = new HashMap<String, Multimap<String, InetAddress>>();
currentLocations = new HashMap<InetAddress, Pair<String, String>>();
dcRacks = new HashMap<>();
currentLocations = new HashMap<>();
}
protected void clear()
void clear()
{
dcEndpoints.clear();
dcRacks.clear();
@ -1143,19 +1175,19 @@ public class TokenMetadata
/**
* construct deep-copy of other
*/
protected Topology(Topology other)
Topology(Topology other)
{
dcEndpoints = HashMultimap.create(other.dcEndpoints);
dcRacks = new HashMap<String, Multimap<String, InetAddress>>();
dcRacks = new HashMap<>();
for (String dc : other.dcRacks.keySet())
dcRacks.put(dc, HashMultimap.create(other.dcRacks.get(dc)));
currentLocations = new HashMap<InetAddress, Pair<String, String>>(other.currentLocations);
currentLocations = new HashMap<>(other.currentLocations);
}
/**
* Stores current DC/rack assignment for ep
*/
protected void addEndpoint(InetAddress ep)
void addEndpoint(InetAddress ep)
{
IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
String dc = snitch.getDatacenter(ep);
@ -1165,10 +1197,14 @@ public class TokenMetadata
{
if (current.left.equals(dc) && current.right.equals(rack))
return;
dcRacks.get(current.left).remove(current.right, ep);
dcEndpoints.remove(current.left, ep);
doRemoveEndpoint(ep, current);
}
doAddEndpoint(ep, dc, rack);
}
private void doAddEndpoint(InetAddress ep, String dc, String rack)
{
dcEndpoints.put(dc, ep);
if (!dcRacks.containsKey(dc))
@ -1181,13 +1217,49 @@ public class TokenMetadata
/**
* Removes current DC/rack assignment for ep
*/
protected void removeEndpoint(InetAddress ep)
void removeEndpoint(InetAddress ep)
{
if (!currentLocations.containsKey(ep))
return;
Pair<String, String> current = currentLocations.remove(ep);
dcEndpoints.remove(current.left, ep);
doRemoveEndpoint(ep, currentLocations.remove(ep));
}
private void doRemoveEndpoint(InetAddress ep, Pair<String, String> current)
{
dcRacks.get(current.left).remove(current.right, ep);
dcEndpoints.remove(current.left, ep);
}
void updateEndpoint(InetAddress ep)
{
IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
if (snitch == null || !currentLocations.containsKey(ep))
return;
updateEndpoint(ep, snitch);
}
void updateEndpoints()
{
IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
if (snitch == null)
return;
for (InetAddress ep : currentLocations.keySet())
updateEndpoint(ep, snitch);
}
private void updateEndpoint(InetAddress ep, IEndpointSnitch snitch)
{
Pair<String, String> current = currentLocations.get(ep);
String dc = snitch.getDatacenter(ep);
String rack = snitch.getRack(ep);
if (dc.equals(current.left) && rack.equals(current.right))
return;
doRemoveEndpoint(ep, current);
doAddEndpoint(ep, dc, rack);
}
/**

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

@ -1681,9 +1681,11 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
SystemKeyspace.updatePeerInfo(endpoint, "release_version", value.value);
break;
case DC:
updateTopology(endpoint);
SystemKeyspace.updatePeerInfo(endpoint, "data_center", value.value);
break;
case RACK:
updateTopology(endpoint);
SystemKeyspace.updatePeerInfo(endpoint, "rack", value.value);
break;
case RPC_ADDRESS:
@ -1711,6 +1713,20 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
}
public void updateTopology(InetAddress endpoint)
{
if (getTokenMetadata().isMember(endpoint))
{
getTokenMetadata().updateTopology(endpoint);
}
}
public void updateTopology()
{
getTokenMetadata().updateTopology();
}
private void updatePeerInfo(InetAddress endpoint)
{
EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint);

213
test/unit/org/apache/cassandra/locator/TokenMetadataTest.java
View File

@ -18,19 +18,29 @@
package org.apache.cassandra.locator;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Map;
import com.google.common.collect.Iterators;
import com.google.common.collect.Multimap;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import static junit.framework.Assert.assertNotNull;
import static org.junit.Assert.assertEquals;
import static org.apache.cassandra.Util.token;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.apache.cassandra.OrderedJUnit4ClassRunner;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
import static org.apache.cassandra.Util.token;
import static org.junit.Assert.assertEquals;
@RunWith(OrderedJUnit4ClassRunner.class)
public class TokenMetadataTest
@ -48,9 +58,9 @@ public class TokenMetadataTest
tmd.updateNormalToken(token(SIX), InetAddress.getByName("127.0.0.6"));
}
private void testRingIterator(ArrayList<Token> ring, String start, boolean includeMin, String... expected)
private static void testRingIterator(ArrayList<Token> ring, String start, boolean includeMin, String... expected)
{
ArrayList<Token> actual = new ArrayList<Token>();
ArrayList<Token> actual = new ArrayList<>();
Iterators.addAll(actual, TokenMetadata.ringIterator(ring, token(start), includeMin));
assertEquals(actual.toString(), expected.length, actual.size());
for (int i = 0; i < expected.length; i++)
@ -82,4 +92,199 @@ public class TokenMetadataTest
{
testRingIterator(new ArrayList<Token>(), "2", false);
}
@Test
public void testTopologyUpdate_RackConsolidation() throws UnknownHostException
{
final InetAddress first = InetAddress.getByName("127.0.0.1");
final InetAddress second = InetAddress.getByName("127.0.0.6");
final String DATA_CENTER = "datacenter1";
final String RACK1 = "rack1";
final String RACK2 = "rack2";
DatabaseDescriptor.setEndpointSnitch(new AbstractEndpointSnitch()
{
@Override
public String getRack(InetAddress endpoint)
{
return endpoint.equals(first) ? RACK1 : RACK2;
}
@Override
public String getDatacenter(InetAddress endpoint)
{
return DATA_CENTER;
}
@Override
public int compareEndpoints(InetAddress target, InetAddress a1, InetAddress a2)
{
return 0;
}
});
tmd.updateNormalToken(token(ONE), first);
tmd.updateNormalToken(token(SIX), second);
TokenMetadata tokenMetadata = tmd.cloneOnlyTokenMap();
assertNotNull(tokenMetadata);
TokenMetadata.Topology topology = tokenMetadata.getTopology();
assertNotNull(topology);
Multimap<String, InetAddress> allEndpoints = topology.getDatacenterEndpoints();
assertNotNull(allEndpoints);
assertTrue(allEndpoints.size() == 2);
assertTrue(allEndpoints.containsKey(DATA_CENTER));
assertTrue(allEndpoints.get(DATA_CENTER).contains(first));
assertTrue(allEndpoints.get(DATA_CENTER).contains(second));
Map<String, Multimap<String, InetAddress>> racks = topology.getDatacenterRacks();
assertNotNull(racks);
assertTrue(racks.size() == 1);
assertTrue(racks.containsKey(DATA_CENTER));
assertTrue(racks.get(DATA_CENTER).size() == 2);
assertTrue(racks.get(DATA_CENTER).containsKey(RACK1));
assertTrue(racks.get(DATA_CENTER).containsKey(RACK2));
assertTrue(racks.get(DATA_CENTER).get(RACK1).contains(first));
assertTrue(racks.get(DATA_CENTER).get(RACK2).contains(second));
DatabaseDescriptor.setEndpointSnitch(new AbstractEndpointSnitch()
{
@Override
public String getRack(InetAddress endpoint)
{
return RACK1;
}
@Override
public String getDatacenter(InetAddress endpoint)
{
return DATA_CENTER;
}
@Override
public int compareEndpoints(InetAddress target, InetAddress a1, InetAddress a2)
{
return 0;
}
});
tokenMetadata.updateTopology(first);
tokenMetadata.updateTopology(second);
allEndpoints = topology.getDatacenterEndpoints();
assertNotNull(allEndpoints);
assertTrue(allEndpoints.size() == 2);
assertTrue(allEndpoints.containsKey(DATA_CENTER));
assertTrue(allEndpoints.get(DATA_CENTER).contains(first));
assertTrue(allEndpoints.get(DATA_CENTER).contains(second));
racks = topology.getDatacenterRacks();
assertNotNull(racks);
assertTrue(racks.size() == 1);
assertTrue(racks.containsKey(DATA_CENTER));
assertTrue(racks.get(DATA_CENTER).size() == 2);
assertTrue(racks.get(DATA_CENTER).containsKey(RACK1));
assertFalse(racks.get(DATA_CENTER).containsKey(RACK2));
assertTrue(racks.get(DATA_CENTER).get(RACK1).contains(first));
assertTrue(racks.get(DATA_CENTER).get(RACK1).contains(second));
}
@Test
public void testTopologyUpdate_RackExpansion() throws UnknownHostException
{
final InetAddress first = InetAddress.getByName("127.0.0.1");
final InetAddress second = InetAddress.getByName("127.0.0.6");
final String DATA_CENTER = "datacenter1";
final String RACK1 = "rack1";
final String RACK2 = "rack2";
DatabaseDescriptor.setEndpointSnitch(new AbstractEndpointSnitch()
{
@Override
public String getRack(InetAddress endpoint)
{
return RACK1;
}
@Override
public String getDatacenter(InetAddress endpoint)
{
return DATA_CENTER;
}
@Override
public int compareEndpoints(InetAddress target, InetAddress a1, InetAddress a2)
{
return 0;
}
});
tmd.updateNormalToken(token(ONE), first);
tmd.updateNormalToken(token(SIX), second);
TokenMetadata tokenMetadata = tmd.cloneOnlyTokenMap();
assertNotNull(tokenMetadata);
TokenMetadata.Topology topology = tokenMetadata.getTopology();
assertNotNull(topology);
Multimap<String, InetAddress> allEndpoints = topology.getDatacenterEndpoints();
assertNotNull(allEndpoints);
assertTrue(allEndpoints.size() == 2);
assertTrue(allEndpoints.containsKey(DATA_CENTER));
assertTrue(allEndpoints.get(DATA_CENTER).contains(first));
assertTrue(allEndpoints.get(DATA_CENTER).contains(second));
Map<String, Multimap<String, InetAddress>> racks = topology.getDatacenterRacks();
assertNotNull(racks);
assertTrue(racks.size() == 1);
assertTrue(racks.containsKey(DATA_CENTER));
assertTrue(racks.get(DATA_CENTER).size() == 2);
assertTrue(racks.get(DATA_CENTER).containsKey(RACK1));
assertFalse(racks.get(DATA_CENTER).containsKey(RACK2));
assertTrue(racks.get(DATA_CENTER).get(RACK1).contains(first));
assertTrue(racks.get(DATA_CENTER).get(RACK1).contains(second));
DatabaseDescriptor.setEndpointSnitch(new AbstractEndpointSnitch()
{
@Override
public String getRack(InetAddress endpoint)
{
return endpoint.equals(first) ? RACK1 : RACK2;
}
@Override
public String getDatacenter(InetAddress endpoint)
{
return DATA_CENTER;
}
@Override
public int compareEndpoints(InetAddress target, InetAddress a1, InetAddress a2)
{
return 0;
}
});
tokenMetadata.updateTopology();
allEndpoints = topology.getDatacenterEndpoints();
assertNotNull(allEndpoints);
assertTrue(allEndpoints.size() == 2);
assertTrue(allEndpoints.containsKey(DATA_CENTER));
assertTrue(allEndpoints.get(DATA_CENTER).contains(first));
assertTrue(allEndpoints.get(DATA_CENTER).contains(second));
racks = topology.getDatacenterRacks();
assertNotNull(racks);
assertTrue(racks.size() == 1);
assertTrue(racks.containsKey(DATA_CENTER));
assertTrue(racks.get(DATA_CENTER).size() == 2);
assertTrue(racks.get(DATA_CENTER).containsKey(RACK1));
assertTrue(racks.get(DATA_CENTER).containsKey(RACK2));
assertTrue(racks.get(DATA_CENTER).get(RACK1).contains(first));
assertTrue(racks.get(DATA_CENTER).get(RACK2).contains(second));
}
}