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.1' into trunk

This commit is contained in:
Sylvain Lebresne 2014-05-01 18:00:31 +02:00
parent c0be34182b e786101eba
commit d8490ccea1
14 changed files with 227 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.rat-excludes
View File

@ -5,6 +5,7 @@ debian/**
**/.project
**/.pydevproject
CHANGES.txt
README.asc
.git/**
**/*.json
**/*.patch

1
CHANGES.txt
View File

@ -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)

5
NEWS.txt
View File

@ -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
---------

2
build.xml
View File

@ -25,7 +25,7 @@
<property name="debuglevel" value="source,lines,vars"/>
<!-- default version and SCM information -->
<property name="base.version" value="2.1.0-beta1"/>
<property name="base.version" value="2.1.0-beta2"/>
<property name="scm.connection" value="scm:git://git.apache.org/cassandra.git"/>
<property name="scm.developerConnection" value="scm:git://git.apache.org/cassandra.git"/>
<property name="scm.url" value="http://git-wip-us.apache.org/repos/asf?p=cassandra.git;a=tree"/>

19
conf/logback-tools.xml
View File

@ -1,3 +1,22 @@
<!--
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.
-->
<configuration>
<appender name="STDERR" target="System.err" class="ch.qos.logback.core.ConsoleAppender">
<encoder>

19
conf/logback.xml
View File

@ -1,3 +1,22 @@
<!--
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.
-->
<configuration scan="true">
<jmxConfigurator />
<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">

6
debian/changelog vendored
View File

@ -1,3 +1,9 @@
cassandra (2.1.0~beta2) unstable; urgency=medium
* New beta release
-- Sylvain Lebresne <slebresne@apache.org> Thu, 01 May 2014 16:39:21 +0200
cassandra (2.1.0~beta1) unstable; urgency=low
* New beta release

2
debian/rules vendored
View File

@ -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

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

@ -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())

81
src/java/org/apache/cassandra/dht/RangeStreamer.java
View File

@ -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<Token> 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<Token> 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<Range<Token>> ranges)
{
Multimap<Range<Token>, InetAddress> rangesForKeyspace = getAllRangesWithSourcesFor(keyspaceName, ranges);
Multimap<Range<Token>, InetAddress> rangesForKeyspace = useStrictConsistency && tokens != null
? getAllRangesWithStrictSourcesFor(keyspaceName, ranges) : getAllRangesWithSourcesFor(keyspaceName, ranges);
if (logger.isDebugEnabled())
{
for (Map.Entry<Range<Token>, InetAddress> entry: rangesForKeyspace.entries())
for (Map.Entry<Range<Token>, 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<Range<Token>, InetAddress> getAllRangesWithStrictSourcesFor(String table, Collection<Range<Token>> desiredRanges)
{
assert tokens != null;
AbstractReplicationStrategy strat = Keyspace.open(table).getReplicationStrategy();
//Active ranges
TokenMetadata metadataClone = metadata.cloneOnlyTokenMap();
Multimap<Range<Token>,InetAddress> addressRanges = strat.getRangeAddresses(metadataClone);
//Pending ranges
metadataClone.updateNormalTokens(tokens, address);
Multimap<Range<Token>,InetAddress> pendingRangeAddresses = strat.getRangeAddresses(metadataClone);
//Collects the source that will have its range moved to the new node
Multimap<Range<Token>, InetAddress> rangeSources = ArrayListMultimap.create();
for (Range<Token> desiredRange : desiredRanges)
{
for (Map.Entry<Range<Token>, Collection<InetAddress>> preEntry : addressRanges.asMap().entrySet())
{
if (preEntry.getKey().contains(desiredRange))
{
Set<InetAddress> oldEndpoints = Sets.newHashSet(preEntry.getValue());
Set<InetAddress> 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<InetAddress> 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

17
src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java
View File

@ -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;

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

@ -3212,7 +3212,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
// getting collection of the currently used ranges by this keyspace
Collection<Range<Token>> currentRanges = getRangesForEndpoint(keyspace, localAddress);
// collection of ranges which this node will serve after move to the new token
Collection<Range<Token>> updatedRanges = strategy.getPendingAddressRanges(tokenMetadata, newToken, localAddress);
Collection<Range<Token>> 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<InetAddress> endpoints = snitch.getSortedListByProximity(localAddress, rangeAddresses.get(range));
List<InetAddress> endpoints = null;
if (RangeStreamer.useStrictConsistency)
{
Set<InetAddress> oldEndpoints = Sets.newHashSet(rangeAddresses.get(range));
Set<InetAddress> 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<InetAddress> 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

19
test/conf/logback-test.xml
View File

@ -1,3 +1,22 @@
<!--
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.
-->
<configuration>
<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>./build/test/logs/system.log</file>

17
tools/stress/src/org/apache/cassandra/stress/settings/OptionCompaction.java
View File

@ -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;