diff --git a/.rat-excludes b/.rat-excludes index 871f5dff62..0da5ab9f97 100644 --- a/.rat-excludes +++ b/.rat-excludes @@ -5,6 +5,7 @@ debian/** **/.project **/.pydevproject CHANGES.txt +README.asc .git/** **/*.json **/*.patch diff --git a/CHANGES.txt b/CHANGES.txt index 3e3a243208..d98174bbf1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -65,6 +65,7 @@ * Optimize cellname comparison (CASSANDRA-6934) * Native protocol v3 (CASSANDRA-6855) * Optimize Cell liveness checks and clean up Cell (CASSANDRA-7119) + * Support consistent range movements (CASSANDRA-2434) Merged from 2.0: * Allow overriding cassandra-rackdc.properties file (CASSANDRA-7072) * Set JMX RMI port to 7199 (CASSANDRA-7087) diff --git a/NEWS.txt b/NEWS.txt index 19503720d0..733c4e72b9 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -39,6 +39,11 @@ New features repair session. Use nodetool repair -par -inc to use this feature. A tool to manually mark/unmark sstables as repaired is available in tools/bin/sstablerepairedset. + - Bootstrapping now ensures that range movements are consistent, + meaning the data for the new node is taken from the node that is no + longer a responsible for that range of keys. + If you want the old behavior (due to a lost node perhaps) + you can set the following property (-Dconsistent.rangemovement=false) Upgrading --------- diff --git a/build.xml b/build.xml index 90fd2589ea..6b74ed5dd9 100644 --- a/build.xml +++ b/build.xml @@ -25,7 +25,7 @@ - + diff --git a/conf/logback-tools.xml b/conf/logback-tools.xml index c472ae483d..ade6c12d86 100644 --- a/conf/logback-tools.xml +++ b/conf/logback-tools.xml @@ -1,3 +1,22 @@ + + diff --git a/conf/logback.xml b/conf/logback.xml index 2657174199..61e5a1341e 100644 --- a/conf/logback.xml +++ b/conf/logback.xml @@ -1,3 +1,22 @@ + + diff --git a/debian/changelog b/debian/changelog index 11f25cb854..4f880a6385 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +cassandra (2.1.0~beta2) unstable; urgency=medium + + * New beta release + + -- Sylvain Lebresne Thu, 01 May 2014 16:39:21 +0200 + cassandra (2.1.0~beta1) unstable; urgency=low * New beta release diff --git a/debian/rules b/debian/rules index 0a43a992d4..f9dfc21393 100755 --- a/debian/rules +++ b/debian/rules @@ -62,7 +62,7 @@ binary-indep: build install dh_testroot dh_installchangelogs dh_installinit -u'start 50 2 3 4 5 . stop 50 0 1 6 .' - dh_installdocs README.txt CHANGES.txt NEWS.txt + dh_installdocs README.asc CHANGES.txt NEWS.txt dh_compress dh_fixperms dh_installdeb diff --git a/src/java/org/apache/cassandra/dht/BootStrapper.java b/src/java/org/apache/cassandra/dht/BootStrapper.java index 343748bbd2..cbbd100cce 100644 --- a/src/java/org/apache/cassandra/dht/BootStrapper.java +++ b/src/java/org/apache/cassandra/dht/BootStrapper.java @@ -63,7 +63,7 @@ public class BootStrapper if (logger.isDebugEnabled()) logger.debug("Beginning bootstrap process"); - RangeStreamer streamer = new RangeStreamer(tokenMetadata, address, "Bootstrap"); + RangeStreamer streamer = new RangeStreamer(tokenMetadata, tokens, address, "Bootstrap"); streamer.addSourceFilter(new RangeStreamer.FailureDetectorSourceFilter(FailureDetector.instance)); for (String keyspaceName : Schema.instance.getNonSystemKeyspaces()) diff --git a/src/java/org/apache/cassandra/dht/RangeStreamer.java b/src/java/org/apache/cassandra/dht/RangeStreamer.java index 7ab39a49d5..2308d304b3 100644 --- a/src/java/org/apache/cassandra/dht/RangeStreamer.java +++ b/src/java/org/apache/cassandra/dht/RangeStreamer.java @@ -23,6 +23,8 @@ import java.util.*; import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.HashMultimap; import com.google.common.collect.Multimap; +import com.google.common.collect.Sets; +import org.apache.cassandra.gms.EndpointState; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,6 +32,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.gms.FailureDetector; +import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.gms.IFailureDetector; import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.locator.IEndpointSnitch; @@ -44,7 +47,8 @@ import org.apache.cassandra.utils.FBUtilities; public class RangeStreamer { private static final Logger logger = LoggerFactory.getLogger(RangeStreamer.class); - + public static final boolean useStrictConsistency = Boolean.valueOf(System.getProperty("consistent.rangemovement","true")); + private final Collection tokens; private final TokenMetadata metadata; private final InetAddress address; private final String description; @@ -99,9 +103,19 @@ public class RangeStreamer } } + public RangeStreamer(TokenMetadata metadata, Collection tokens, InetAddress address, String description) + { + this.metadata = metadata; + this.tokens = tokens; + this.address = address; + this.description = description; + this.streamPlan = new StreamPlan(description); + } + public RangeStreamer(TokenMetadata metadata, InetAddress address, String description) { this.metadata = metadata; + this.tokens = null; this.address = address; this.description = description; this.streamPlan = new StreamPlan(description); @@ -114,11 +128,12 @@ public class RangeStreamer public void addRanges(String keyspaceName, Collection> ranges) { - Multimap, InetAddress> rangesForKeyspace = getAllRangesWithSourcesFor(keyspaceName, ranges); + Multimap, InetAddress> rangesForKeyspace = useStrictConsistency && tokens != null + ? getAllRangesWithStrictSourcesFor(keyspaceName, ranges) : getAllRangesWithSourcesFor(keyspaceName, ranges); if (logger.isDebugEnabled()) { - for (Map.Entry, InetAddress> entry: rangesForKeyspace.entries()) + for (Map.Entry, InetAddress> entry : rangesForKeyspace.entries()) logger.debug(String.format("%s: range %s exists on %s", description, entry.getKey(), entry.getValue())); } @@ -162,6 +177,66 @@ public class RangeStreamer return rangeSources; } + /** + * Get a map of all ranges and the source that will be cleaned up once this bootstrapped node is added for the given ranges. + * For each range, the list should only contain a single source. This allows us to consistently migrate data without violating + * consistency. + */ + private Multimap, InetAddress> getAllRangesWithStrictSourcesFor(String table, Collection> desiredRanges) + { + + assert tokens != null; + AbstractReplicationStrategy strat = Keyspace.open(table).getReplicationStrategy(); + + //Active ranges + TokenMetadata metadataClone = metadata.cloneOnlyTokenMap(); + Multimap,InetAddress> addressRanges = strat.getRangeAddresses(metadataClone); + + //Pending ranges + metadataClone.updateNormalTokens(tokens, address); + Multimap,InetAddress> pendingRangeAddresses = strat.getRangeAddresses(metadataClone); + + //Collects the source that will have its range moved to the new node + Multimap, InetAddress> rangeSources = ArrayListMultimap.create(); + + for (Range desiredRange : desiredRanges) + { + for (Map.Entry, Collection> preEntry : addressRanges.asMap().entrySet()) + { + if (preEntry.getKey().contains(desiredRange)) + { + Set oldEndpoints = Sets.newHashSet(preEntry.getValue()); + Set newEndpoints = Sets.newHashSet(pendingRangeAddresses.get(desiredRange)); + + //Due to CASSANDRA-5953 we can have a higher RF then we have endpoints. + //So we need to be careful to only be strict when endpoints == RF + if (oldEndpoints.size() == strat.getReplicationFactor()) + { + oldEndpoints.removeAll(newEndpoints); + assert oldEndpoints.size() == 1 : "Expected 1 endpoint but found " + oldEndpoints.size(); + } + + rangeSources.put(desiredRange, oldEndpoints.iterator().next()); + } + } + + //Validate + Collection addressList = rangeSources.get(desiredRange); + if (addressList == null || addressList.isEmpty()) + throw new IllegalStateException("No sources found for " + desiredRange); + + if (addressList.size() > 1) + throw new IllegalStateException("Multiple endpoints found for " + desiredRange); + + InetAddress sourceIp = addressList.iterator().next(); + EndpointState sourceState = Gossiper.instance.getEndpointStateForEndpoint(sourceIp); + if (Gossiper.instance.isEnabled() && (sourceState == null || !sourceState.isAlive())) + throw new RuntimeException("A node required to move the data consistently is down ("+sourceIp+"). If you wish to move the data from a potentially inconsistent replica, restart the node with -Dconsistent.rangemovement=false"); + } + + return rangeSources; + } + /** * @param rangesWithSources The ranges we want to fetch (key) and their potential sources (value) * @param sourceFilters A (possibly empty) collection of source filters to apply. In addition to any filters given diff --git a/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java b/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java index 98492da262..b95bf32bcf 100644 --- a/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java +++ b/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java @@ -1,3 +1,20 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.apache.cassandra.io.util; import java.io.File; diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index d94c0cd0df..34bc422bb9 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -3212,7 +3212,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE // getting collection of the currently used ranges by this keyspace Collection> currentRanges = getRangesForEndpoint(keyspace, localAddress); // collection of ranges which this node will serve after move to the new token - Collection> updatedRanges = strategy.getPendingAddressRanges(tokenMetadata, newToken, localAddress); + Collection> updatedRanges = strategy.getPendingAddressRanges(tokenMetaClone, newToken, localAddress); // ring ranges and endpoints associated with them // this used to determine what nodes should we ping about range data @@ -3232,11 +3232,51 @@ public class StorageService extends NotificationBroadcasterSupport implements IE { if (range.contains(toFetch)) { - List endpoints = snitch.getSortedListByProximity(localAddress, rangeAddresses.get(range)); + List endpoints = null; + + if (RangeStreamer.useStrictConsistency) + { + Set oldEndpoints = Sets.newHashSet(rangeAddresses.get(range)); + Set newEndpoints = Sets.newHashSet(strategy.calculateNaturalEndpoints(toFetch.right, tokenMetaCloneAllSettled)); + + //Due to CASSANDRA-5953 we can have a higher RF then we have endpoints. + //So we need to be careful to only be strict when endpoints == RF + if (oldEndpoints.size() == strategy.getReplicationFactor()) + { + oldEndpoints.removeAll(newEndpoints); + + //No relocation required + if (oldEndpoints.isEmpty()) + continue; + + assert oldEndpoints.size() == 1 : "Expected 1 endpoint but found " + oldEndpoints.size(); + } + + endpoints = Lists.newArrayList(oldEndpoints.iterator().next()); + } + else + { + endpoints = snitch.getSortedListByProximity(localAddress, rangeAddresses.get(range)); + } + // storing range and preferred endpoint set rangesToFetchWithPreferredEndpoints.putAll(toFetch, endpoints); } } + + Collection addressList = rangesToFetchWithPreferredEndpoints.get(toFetch); + if (addressList == null || addressList.isEmpty()) + continue; + + if (RangeStreamer.useStrictConsistency) + { + if (addressList.size() > 1) + throw new IllegalStateException("Multiple strict sources found for " + toFetch); + + InetAddress sourceIp = addressList.iterator().next(); + if (Gossiper.instance.isEnabled() && !Gossiper.instance.getEndpointStateForEndpoint(sourceIp).isAlive()) + throw new RuntimeException("A node required to move the data consistently is down ("+sourceIp+"). If you wish to move the data from a potentially inconsistent replica, restart the node with -Dconsistent.rangemovement=false"); + } } // calculating endpoints to stream current ranges to if needed diff --git a/test/conf/logback-test.xml b/test/conf/logback-test.xml index 89caa2f886..a7aa34c63f 100644 --- a/test/conf/logback-test.xml +++ b/test/conf/logback-test.xml @@ -1,3 +1,22 @@ + + ./build/test/logs/system.log diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionCompaction.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionCompaction.java index da74e430e8..11d5403a19 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionCompaction.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionCompaction.java @@ -1,3 +1,20 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.apache.cassandra.stress.settings; import java.util.Arrays;