diff --git a/CHANGES.txt b/CHANGES.txt index 36ae689c36..4b2ce1e64a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java b/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java index 2c0980a7df..da43600bb4 100644 --- a/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java +++ b/src/java/org/apache/cassandra/locator/GossipingPropertyFileSnitch.java @@ -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); diff --git a/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java b/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java index 8665816b51..98262c4c41 100644 --- a/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java +++ b/src/java/org/apache/cassandra/locator/PropertyFileSnitch.java @@ -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); diff --git a/src/java/org/apache/cassandra/locator/TokenMetadata.java b/src/java/org/apache/cassandra/locator/TokenMetadata.java index c3ec353653..75448a8616 100644 --- a/src/java/org/apache/cassandra/locator/TokenMetadata.java +++ b/src/java/org/apache/cassandra/locator/TokenMetadata.java @@ -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 bootstrapTokens = new BiMultiValMap(); + private final BiMultiValMap 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 leavingEndpoints = new HashSet(); + private final Set leavingEndpoints = new HashSet<>(); // this is a cache of the calculation from {tokenToEndpointMap, bootstrapTokens, leavingEndpoints} - private final ConcurrentMap, InetAddress>> pendingRanges = new ConcurrentHashMap, InetAddress>>(); + private final ConcurrentMap, InetAddress>> pendingRanges = new ConcurrentHashMap<>(); // nodes which are migrating to the new tokens in the ring - private final Set> movingEndpoints = new HashSet>(); + private final Set> 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 sortTokens() { - return new ArrayList(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 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 readMap = new HashMap(); + Map 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(tokenToEndpointMap.inverse().get(endpoint)); + return new ArrayList<>(tokenToEndpointMap.inverse().get(endpoint)); } finally { @@ -522,7 +554,7 @@ public class TokenMetadata } } - private final AtomicReference cachedTokenMap = new AtomicReference(); + private final AtomicReference 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.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> getPrimaryRangesFor(Collection tokens) { - Collection> ranges = new ArrayList>(tokens.size()); + Collection> ranges = new ArrayList<>(tokens.size()); for (Token right : tokens) - ranges.add(new Range(getPredecessor(right), right)); + ranges.add(new Range<>(getPredecessor(right), right)); return ranges; } @@ -675,7 +707,7 @@ public class TokenMetadata public List> getPendingRanges(String keyspaceName, InetAddress endpoint) { - List> ranges = new ArrayList>(); + List> ranges = new ArrayList<>(); for (Map.Entry, 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 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, 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 endpoints = new HashSet(); + Set endpoints = new HashSet<>(); for (Map.Entry, Collection> entry : ranges.entrySet()) { if (entry.getKey().contains(token)) @@ -1074,7 +1106,7 @@ public class TokenMetadata lock.readLock().lock(); try { - Map map = new HashMap(tokenToEndpointMap.size() + bootstrapTokens.size()); + Map 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> currentLocations; - protected Topology() + Topology() { dcEndpoints = HashMultimap.create(); - dcRacks = new HashMap>(); - currentLocations = new HashMap>(); + 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>(); + dcRacks = new HashMap<>(); for (String dc : other.dcRacks.keySet()) dcRacks.put(dc, HashMultimap.create(other.dcRacks.get(dc))); - currentLocations = new HashMap>(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 current = currentLocations.remove(ep); - dcEndpoints.remove(current.left, ep); + + doRemoveEndpoint(ep, currentLocations.remove(ep)); + } + + private void doRemoveEndpoint(InetAddress ep, Pair 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 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); } /** diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index fc2153f699..f09563015e 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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); diff --git a/test/unit/org/apache/cassandra/locator/TokenMetadataTest.java b/test/unit/org/apache/cassandra/locator/TokenMetadataTest.java index 68d0dfa4b8..e7bb70a78b 100644 --- a/test/unit/org/apache/cassandra/locator/TokenMetadataTest.java +++ b/test/unit/org/apache/cassandra/locator/TokenMetadataTest.java @@ -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 ring, String start, boolean includeMin, String... expected) + private static void testRingIterator(ArrayList ring, String start, boolean includeMin, String... expected) { - ArrayList actual = new ArrayList(); + ArrayList 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(), "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 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> 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 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> 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)); + } }