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

merge from 0.7 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1043262 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-12-08 01:08:32 +00:00
parent c5d5088f96 3a63f3e2e0
commit 4fe8af8ea3
38 changed files with 471 additions and 715 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
CHANGES.txt
View File

@ -1,4 +1,13 @@
dev
0.8-dev
* avoid double RowMutation serialization on write path (CASSANDRA-1800)
0.7-dev
* expose getNaturalEndpoints in StorageServiceMBean taking byte[]
key; RMI cannot serialize ByteBuffer (CASSANDRA-1833)
0.7.0-rc2
* fix live-column-count of slice ranges including tombstoned supercolumn
with live subcolumn (CASSANDRA-1591)
* rename o.a.c.internal.AntientropyStage -> AntiEntropyStage,
@ -23,7 +32,17 @@ dev
* close file handle used for post-flush truncate (CASSANDRA-1790)
* various code cleanup (CASSANDRA-1793, -1794, -1795)
* fix range queries against wrapped range (CASSANDRA-1781)
* avoid double RowMutation serialization on write path (CASSANDRA-1800)
* fix consistencylevel calculations for NetworkTopologyStrategy
(CASSANDRA-1804)
* cli support index type enum names (CASSANDRA-1810)
* improved validation of column_metadata (CASSANDRA-1813)
* reads at ConsistencyLevel > 1 throw UnavailableException
immediately if insufficient live nodes exist (CASSANDRA-1803)
* copy bytebuffers for local writes to avoid retaining the entire
Thrift frame (CASSANDRA-1801)
* fix NPE adding index to column w/o prior metadata (CASSANDRA-1764)
* reduce fat client timeout (CASSANDRA-1730)
* fix botched merge of CASSANDRA-1316
0.7.0-rc1

2
build.xml
View File

@ -47,7 +47,7 @@
<property name="test.unit.src" value="${test.dir}/unit"/>
<property name="test.long.src" value="${test.dir}/long"/>
<property name="dist.dir" value="${build.dir}/dist"/>
<property name="base.version" value="0.7.0-beta3"/>
<property name="base.version" value="0.7.0-rc2"/>
<condition property="version" value="${base.version}">
<isset property="release"/>
</condition>

10
conf/cassandra-env.sh
View File

@ -78,7 +78,7 @@ if [ "`uname`" = "Linux" ] ; then
JVM_OPTS="$JVM_OPTS -Xss128k"
fi
# GC tuning options.
# GC tuning options
JVM_OPTS="$JVM_OPTS -XX:+UseParNewGC"
JVM_OPTS="$JVM_OPTS -XX:+UseConcMarkSweepGC"
JVM_OPTS="$JVM_OPTS -XX:+CMSParallelRemarkEnabled"
@ -87,6 +87,14 @@ JVM_OPTS="$JVM_OPTS -XX:MaxTenuringThreshold=1"
JVM_OPTS="$JVM_OPTS -XX:CMSInitiatingOccupancyFraction=75"
JVM_OPTS="$JVM_OPTS -XX:+UseCMSInitiatingOccupancyOnly"
# GC logging options -- uncomment to enable
# JVM_OPTS="$JVM_OPTS -XX:+PrintGCDetails"
# JVM_OPTS="$JVM_OPTS -XX:+PrintGCTimeStamps"
# JVM_OPTS="$JVM_OPTS -XX:+PrintClassHistogram"
# JVM_OPTS="$JVM_OPTS -XX:+PrintTenuringDistribution"
# JVM_OPTS="$JVM_OPTS -XX:+PrintGCApplicationStoppedTime"
# JVM_OPTS="$JVM_OPTS -Xloggc:/var/log/cassandra/gc.log"
# Prefer binding to IPv4 network intefaces (when net.ipv6.bindv6only=1). See
# http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6342561 (short version:
# comment out this entry to enable IPv6 support).

551
contrib/maven/pom.xml
View File

@ -1,551 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<parent>
<groupId>org.apache</groupId>
<artifactId>apache</artifactId>
<version>6</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>org.apache.cassandra</groupId>
<artifactId>cassandra</artifactId>
<version>0.6-SNAPSHOT</version>
<packaging>jar</packaging>
<name>Cassandra</name>
<inceptionYear>2009</inceptionYear>
<prerequisites>
<maven>2.0.9</maven>
</prerequisites>
<url>http://incubator.apache.org/cassandra</url>
<mailingLists>
<mailingList>
<name>cassandra-user</name>
<subscribe>cassandra-user-subscribe@incubator.apache.org</subscribe>
<unsubscribe>cassandra-user-unsubscribe@incubator.apache.org</unsubscribe>
<post>cassandra-user@incubator.apache.org</post>
<archive>http://mail-archives.apache.org/mod_mbox/incubator-cassandra-user/</archive>
</mailingList>
<mailingList>
<name>cassandra-dev</name>
<subscribe>cassandra-dev-subscribe@incubator.apache.org</subscribe>
<unsubscribe>cassandra-dev-unsubscribe@incubator.apache.org</unsubscribe>
<post>cassandra-dev@incubator.apache.org</post>
<archive>http://mail-archives.apache.org/mod_mbox/incubator-cassandra-dev/</archive>
</mailingList>
<mailingList>
<name>cassandra-commits</name>
<subscribe>cassandra-commits-subscribe@incubator.apache.org</subscribe>
<unsubscribe>cassandra-commits-unsubscribe@incubator.apache.org</unsubscribe>
<post>cassandra-commits@incubator.apache.org</post>
<archive>http://mail-archives.apache.org/mod_mbox/incubator-cassandra-commits/</archive>
</mailingList>
</mailingLists>
<!-- This section intentionally left blank (see: CASSANDRA-415) -->
<developers>
<developer>
</developer>
</developers>
<issueManagement>
<system>JIRA</system>
<url>https://issues.apache.org/jira/browse/CASSANDRA</url>
</issueManagement>
<!-- see: https://hudson.dev.java.net/maven-hudson-plugin (to kick off hudson jobs via mvn) -->
<ciManagement>
<system>hudson</system>
<url>http://hudson.zones.apache.org/hudson</url>
<notifiers>
<!-- Hopefully one day hudson will automatically load this data -->
<notifier>
<type>mail</type>
<sendOnError>true</sendOnError>
<sendOnFailure>true</sendOnFailure>
<sendOnSuccess>false</sendOnSuccess>
<sendOnWarning>false</sendOnWarning>
<configuration><address>cassandra-commits@incubator.apache.org</address></configuration>
</notifier>
</notifiers>
</ciManagement>
<distributionManagement>
<site>
<id>cassandra-website</id>
<url>scp://people.apache.org/x1/www/incubator.apache.org/cassandra/maven/${pom.version}</url>
</site>
</distributionManagement>
<scm>
<connection>scm:svn:http://svn.apache.org/repos/asf/incubator/cassandra/trunk</connection>
<developerConnection>scm:svn:https://svn.apache.org/repos/asf/incubator/cassandra/trunk</developerConnection>
<url>http://svn.apache.org/viewvc/incubator/cassandra/trunk/</url>
</scm>
<dependencies>
<dependency>
<groupId>commons-collections</groupId>
<artifactId>commons-collections</artifactId>
<version>3.2.1</version>
</dependency>
<dependency>
<groupId>commons-cli</groupId>
<artifactId>commons-cli</artifactId>
<version>1.1</version>
</dependency>
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.4</version>
</dependency>
<dependency>
<groupId>jline</groupId>
<artifactId>jline</artifactId>
<version>0.9.94</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.15</version>
<!-- And the exclusion below illustrate why we should try to make as many of our jars optional -->
<exclusions>
<exclusion>
<groupId>javax.jms</groupId>
<artifactId>jms</artifactId>
</exclusion>
<exclusion>
<groupId>com.sun.jmx</groupId>
<artifactId>jmxri</artifactId>
</exclusion>
<exclusion>
<groupId>com.sun.jdmk</groupId>
<artifactId>jmxtools</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.5.8</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.5.8</version>
</dependency>
<dependency>
<groupId>org.antlr</groupId>
<artifactId>antlr-runtime</artifactId>
<version>3.1.3</version>
</dependency>
<dependency>
<groupId>com.google.collections</groupId>
<artifactId>google-collections</artifactId>
<version>1.0-rc1</version>
</dependency>
<!-- Dependencies which I could not find in the maven repos -->
<dependency>
<groupId>high-scale-lib</groupId>
<artifactId>high-scale-lib</artifactId>
<version>UNKNOWN</version>
<scope>system</scope>
<systemPath>${basedir}/lib/high-scale-lib.jar</systemPath>
</dependency>
<dependency>
<groupId>flexjson</groupId>
<artifactId>flexjson</artifactId>
<version>1.7</version>
<scope>system</scope>
<systemPath>${basedir}/lib/flexjson-1.7.jar</systemPath>
</dependency>
<dependency>
<groupId>libthrift</groupId>
<artifactId>libthrift</artifactId>
<version>UNKNOWN</version>
<scope>system</scope>
<systemPath>${basedir}/lib/libthrift-r820831.jar</systemPath>
</dependency>
<dependency>
<groupId>jsonsimple</groupId>
<artifactId>jsonsimple</artifactId>
<version>UNKNOWN</version>
<scope>system</scope>
<systemPath>${basedir}/lib/json_simple-1.1.jar</systemPath>
</dependency>
<dependency>
<groupId>com.reardencommerce</groupId>
<artifactId>clhm</artifactId>
<version>UNKNOWN</version>
<scope>system</scope>
<systemPath>${basedir}/lib/clhm-production.jar</systemPath>
</dependency>
<!-- Testing Dependencies -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.6</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<sourceDirectory>${basedir}/src/java</sourceDirectory>
<testSourceDirectory>${basedir}/test/unit</testSourceDirectory>
<outputDirectory>build/classes</outputDirectory>
<testResources>
<testResource>
<directory>${basedir}/test/conf</directory>
<includes>
<include>**/*</include>
</includes>
</testResource>
<testResource>
<directory>${basedir}/test/resources</directory>
<includes>
<include>*.json</include>
</includes>
</testResource>
</testResources>
<plugins>
<!-- builds the antlr sources -->
<plugin>
<groupId>org.antlr</groupId>
<artifactId>antlr3-maven-plugin</artifactId>
<version>3.1.3-1</version>
<executions>
<execution>
<phase>process-sources</phase>
<goals>
<goal>antlr</goal>
</goals>
<configuration>
<sourceDirectory>${basedir}/src/java</sourceDirectory>
</configuration>
</execution>
</executions>
</plugin>
<!-- Include the thrift generated sources in the build -->
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>build-helper-maven-plugin</artifactId>
<version>1.3</version>
<executions>
<execution>
<id>add-source</id>
<phase>generate-sources</phase>
<goals>
<goal>add-source</goal>
</goals>
<configuration>
<sources>
<source>${basedir}/interface/gen-java</source>
</sources>
</configuration>
</execution>
</executions>
</plugin>
<!-- Setup the build for JAVA 1.6 -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.6</source>
<target>1.6</target>
<optimize>true</optimize>
<debug>true</debug>
<showDeprecation>true</showDeprecation>
<showWarnings>true</showWarnings>
</configuration>
</plugin>
<!-- Configure the JUnit Env -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<systemProperties>
<property>
<name>storage-config</name>
<value>${basedir}/test/conf</value>
</property>
</systemProperties>
<forkMode>always</forkMode>
<excludes>
<exclude>**/TestRingCache.java</exclude>
</excludes>
</configuration>
</plugin>
<!-- Lets run with cobertura -->
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>cobertura-maven-plugin</artifactId>
<version>2.0</version>
<configuration>
</configuration>
</plugin>
<!-- These plugins help validate ASF legal requirements are met for the release -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-release-plugin</artifactId>
<version>2.0-beta-9</version>
<configuration>
<autoVersionSubmodules>true</autoVersionSubmodules>
<allowTimestampedSnapshots>false</allowTimestampedSnapshots>
<preparationGoals>clean install</preparationGoals>
<goals>deploy</goals>
<arguments>-Papache-release</arguments>
</configuration>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>ianal-maven-plugin</artifactId>
<version>1.0-alpha-1</version>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>rat-maven-plugin</artifactId>
<version>1.0-alpha-3</version>
<configuration>
<excludeSubProjects>false</excludeSubProjects>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-enforcer-plugin</artifactId>
<executions>
<execution>
<phase>validate</phase>
<goals>
<goal>enforce</goal>
</goals>
<configuration>
<rules>
<requireMavenVersion>
<version>[2.0.9,)</version>
</requireMavenVersion>
</rules>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>ianal-maven-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>verify-legal-files</goal>
</goals>
<configuration>
<strict>true</strict>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<reporting>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jxr-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-report-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-pmd-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>taglist-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-javadoc-plugin</artifactId>
<configuration>
<links>
<link>http://java.sun.com/j2se/1.6.0/docs/api/</link>
<link>http://logging.apache.org/log4j/docs/api/</link>
</links>
<!--
<stylesheetfile>${basedir}/../etc/css/stylesheet.css</stylesheetfile>
-->
<linksource>true</linksource>
<maxmemory>900m</maxmemory>
<source>1.6</source>
</configuration>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>cobertura-maven-plugin</artifactId>
<version>2.2</version>
<configuration>
<formats>
<format>html</format>
<format>xml</format>
</formats>
</configuration>
</plugin>
</plugins>
</reporting>
<profiles>
<!-- run "mvn -P thrift" to generate the thrift classes -->
<profile>
<id>thrift</id>
<build>
<defaultGoal>process-sources</defaultGoal>
<plugins>
<plugin>
<artifactId>maven-antrun-plugin</artifactId>
<executions>
<execution>
<phase>process-sources</phase>
<configuration>
<tasks>
<ant antfile="${basedir}/build.xml" target="gen-thrift-java"/>
<ant antfile="${basedir}/build.xml" target="gen-thrift-py"/>
</tasks>
</configuration>
<goals>
<goal>run</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>apache-release</id>
<build>
<plugins>
<!-- We want to sign the artifact, the POM, and all attached artifacts -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-gpg-plugin</artifactId>
<configuration>
<passphrase>${gpg.passphrase}</passphrase>
</configuration>
<executions>
<execution>
<goals>
<goal>sign</goal>
</goals>
</execution>
</executions>
</plugin>
<!-- We want to deploy the artifact to a staging location for perusal -->
<plugin>
<inherited>true</inherited>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-deploy-plugin</artifactId>
<configuration>
<updateReleaseInfo>true</updateReleaseInfo>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-source-plugin</artifactId>
<executions>
<execution>
<id>attach-sources</id>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-javadoc-plugin</artifactId>
<configuration>
<encoding>${project.build.sourceEncoding}</encoding>
</configuration>
<executions>
<execution>
<id>attach-javadocs</id>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
</plugin>
<!-- Deploy a source distro too -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>single</goal>
</goals>
<phase>package</phase>
<configuration>
<runOnlyAtExecutionRoot>true</runOnlyAtExecutionRoot>
<descriptorRefs>
<descriptorRef>
source-release
</descriptorRef>
</descriptorRefs>
</configuration>
</execution>
</executions>
<dependencies>
<dependency>
<groupId>org.apache.geronimo.genesis</groupId>
<artifactId>apache-source-release-assembly-descriptor</artifactId>
<version>2.0</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
</profile>
</profiles>
</project>

6
debian/changelog vendored
View File

@ -1,3 +1,9 @@
cassandra (0.7.0~rc2) unstable; urgency=low
* Release candidate release.
-- Eric Evans <eevans@apache.org> Mon, 06 Dec 2010 11:19:40 -0600
cassandra (0.7.0~rc1) unstable; urgency=low
* Release candidate release.

28
src/java/org/apache/cassandra/cli/CliClient.java
View File

@ -468,7 +468,7 @@ public class CliClient extends CliUserHelp
return;
IndexClause clause = new IndexClause();
String columnFamily = statement.getChild(0).getText();
String columnFamily = CliCompiler.getColumnFamily(statement, keyspacesMap.get(keySpace).cf_defs);
// ^(CONDITIONS ^(CONDITION $column $value) ...)
Tree conditions = statement.getChild(1);
@ -1414,20 +1414,28 @@ public class CliClient extends CliUserHelp
*/
private IndexType getIndexTypeFromString(String indexTypeAsString)
{
Integer indexTypeId;
IndexType indexType;
try {
indexTypeId = new Integer(indexTypeAsString);
try
{
indexType = IndexType.findByValue(new Integer(indexTypeAsString));
}
catch (NumberFormatException e) {
throw new RuntimeException("Could not convert " + indexTypeAsString + " into Integer.");
catch (NumberFormatException e)
{
try
{
// if this is not an integer lets try to get IndexType by name
indexType = IndexType.valueOf(indexTypeAsString);
}
catch (IllegalArgumentException ie)
{
throw new RuntimeException("IndexType '" + indexTypeAsString + "' is unsupported.");
}
}
indexType = IndexType.findByValue(indexTypeId);
if (indexType == null) {
throw new RuntimeException(indexTypeAsString + " is unsupported.");
if (indexType == null)
{
throw new RuntimeException("IndexType '" + indexTypeAsString + "' is unsupported.");
}
return indexType;

5
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -41,6 +41,7 @@ import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.migration.Migration;
import org.apache.cassandra.io.SerDeUtils;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.commons.lang.builder.EqualsBuilder;
@ -751,7 +752,7 @@ public final class CFMetaData
org.apache.cassandra.avro.ColumnDef tcd = new org.apache.cassandra.avro.ColumnDef();
tcd.index_name = cd.getIndexName();
tcd.index_type = org.apache.cassandra.avro.IndexType.valueOf(cd.getIndexType().name());
tcd.name = cd.name;
tcd.name = ByteBufferUtil.clone(cd.name);
tcd.validation_class = cd.validator.getClass().getName();
column_meta.add(tcd);
}
@ -786,7 +787,7 @@ public final class CFMetaData
for (org.apache.cassandra.thrift.ColumnDef cdef : def.getColumn_metadata())
{
org.apache.cassandra.avro.ColumnDef tdef = new org.apache.cassandra.avro.ColumnDef();
tdef.name = cdef.BufferForName();
tdef.name = ByteBufferUtil.clone(cdef.BufferForName());
tdef.validation_class = cdef.getValidation_class();
tdef.index_name = cdef.getIndex_name();
tdef.index_type = cdef.getIndex_type() == null ? null : org.apache.cassandra.avro.IndexType.valueOf(cdef.getIndex_type().name());

15
src/java/org/apache/cassandra/config/ColumnDefinition.java
View File

@ -31,6 +31,7 @@ import org.apache.avro.util.Utf8;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.thrift.ColumnDef;
import org.apache.cassandra.thrift.IndexType;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
public class ColumnDefinition {
@ -103,8 +104,7 @@ public class ColumnDefinition {
public static ColumnDefinition fromColumnDef(ColumnDef thriftColumnDef) throws ConfigurationException
{
validateIndexType(thriftColumnDef);
return new ColumnDefinition(thriftColumnDef.name, thriftColumnDef.validation_class, thriftColumnDef.index_type, thriftColumnDef.index_name);
return new ColumnDefinition(ByteBufferUtil.clone(thriftColumnDef.name), thriftColumnDef.validation_class, thriftColumnDef.index_type, thriftColumnDef.index_name);
}
public static ColumnDefinition fromColumnDef(org.apache.cassandra.avro.ColumnDef avroColumnDef) throws ConfigurationException
@ -123,10 +123,7 @@ public class ColumnDefinition {
Map<ByteBuffer, ColumnDefinition> cds = new TreeMap<ByteBuffer, ColumnDefinition>();
for (ColumnDef thriftColumnDef : thriftDefs)
{
validateIndexType(thriftColumnDef);
cds.put(thriftColumnDef.name, fromColumnDef(thriftColumnDef));
}
cds.put(ByteBufferUtil.clone(thriftColumnDef.name), fromColumnDef(thriftColumnDef));
return Collections.unmodifiableMap(cds);
}
@ -146,12 +143,6 @@ public class ColumnDefinition {
return Collections.unmodifiableMap(cds);
}
public static void validateIndexType(org.apache.cassandra.thrift.ColumnDef thriftColumnDef) throws ConfigurationException
{
if ((thriftColumnDef.index_name != null) && (thriftColumnDef.index_type == null))
throw new ConfigurationException("index_name cannot be set if index_type is not also set");
}
public static void validateIndexType(org.apache.cassandra.avro.ColumnDef avroColumnDef) throws ConfigurationException
{
if ((avroColumnDef.index_name != null) && (avroColumnDef.index_type == null))

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

@ -816,11 +816,6 @@ public class DatabaseDescriptor
return conf.rpc_port;
}
public static int getReplicationFactor(String table)
{
return tables.get(table).replicationFactor;
}
public static long getRpcTimeout()
{
return conf.rpc_timeout_in_ms;

6
src/java/org/apache/cassandra/db/Column.java
View File

@ -210,6 +210,12 @@ public class Column implements IColumn
return result;
}
@Override
public IColumn deepCopy()
{
return new Column(ByteBufferUtil.clone(name), ByteBufferUtil.clone(value), timestamp);
}
public String getString(AbstractType comparator)
{
StringBuilder sb = new StringBuilder();

5
src/java/org/apache/cassandra/db/ColumnFamilyType.java
View File

@ -25,11 +25,12 @@ public enum ColumnFamilyType
Standard,
Super;
public final static ColumnFamilyType create(String name)
public static ColumnFamilyType create(String name)
{
try
{
return name == null ? null : ColumnFamilyType.valueOf(name);
// TODO thrift optional parameter in CfDef is leaking down here which it shouldn't
return name == null ? ColumnFamilyType.Standard : ColumnFamilyType.valueOf(name);
}
catch (IllegalArgumentException e)
{

7
src/java/org/apache/cassandra/db/DeletedColumn.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.db;
import java.nio.ByteBuffer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -55,4 +56,10 @@ public class DeletedColumn extends Column
{
return value.getInt(value.position()+value.arrayOffset() );
}
@Override
public IColumn deepCopy()
{
return new DeletedColumn(ByteBufferUtil.clone(name), ByteBufferUtil.clone(value), timestamp);
}
}

7
src/java/org/apache/cassandra/db/ExpiringColumn.java
View File

@ -24,6 +24,7 @@ import java.security.MessageDigest;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.log4j.Logger;
/**
@ -104,6 +105,12 @@ public class ExpiringColumn extends Column
return localExpirationTime;
}
@Override
public IColumn deepCopy()
{
return new ExpiringColumn(ByteBufferUtil.clone(name), ByteBufferUtil.clone(value), timestamp, timeToLive, localExpirationTime);
}
@Override
public String getString(AbstractType comparator)
{

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

@ -24,6 +24,7 @@ import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;
@ -122,7 +123,7 @@ public class HintedHandOffManager
rm.add(cf);
Message message = rm.makeRowMutationMessage();
IWriteResponseHandler responseHandler = WriteResponseHandler.create(endpoint);
MessagingService.instance.sendRR(message, new InetAddress[] { endpoint }, responseHandler);
MessagingService.instance.sendRR(message, Arrays.asList(endpoint), responseHandler);
try
{
responseHandler.get();

3
src/java/org/apache/cassandra/db/IColumn.java
View File

@ -46,6 +46,9 @@ public interface IColumn
public int getLocalDeletionTime(); // for tombstone GC, so int is sufficient granularity
public String getString(AbstractType comparator);
/** clones the column, making copies of any underlying byte buffers */
IColumn deepCopy();
/**
* For a simple column, live == !isMarkedForDelete.
* For a supercolumn, live means it has at least one subcolumn whose timestamp is greater than the

16
src/java/org/apache/cassandra/db/RowMutation.java
View File

@ -41,6 +41,7 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.Deletion;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.commons.lang.StringUtils;
@ -317,6 +318,21 @@ public class RowMutation
rm.preserializedBuffer = raw;
return rm;
}
public RowMutation deepCopy()
{
RowMutation rm = new RowMutation(table_, ByteBufferUtil.clone(key_));
for (Map.Entry<Integer, ColumnFamily> e : modifications_.entrySet())
{
ColumnFamily cf = e.getValue().cloneMeShallow();
for (Map.Entry<ByteBuffer, IColumn> ce : e.getValue().getColumnsMap().entrySet())
cf.addColumn(ce.getValue().deepCopy());
rm.modifications_.put(e.getKey(), cf);
}
return rm;
}
}
class RowMutationSerializer implements ICompactSerializer<RowMutation>

16
src/java/org/apache/cassandra/db/SuperColumn.java
View File

@ -24,6 +24,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@ -31,6 +32,7 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.io.ICompactSerializer2;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -286,6 +288,20 @@ public class SuperColumn implements IColumn, IColumnContainer
this.localDeletionTime.set(localDeleteTime);
this.markedForDeleteAt.set(timestamp);
}
public IColumn deepCopy()
{
SuperColumn sc = new SuperColumn(ByteBufferUtil.clone(name_), this.getComparator());
sc.localDeletionTime = localDeletionTime;
sc.markedForDeleteAt = markedForDeleteAt;
for(Map.Entry<ByteBuffer, IColumn> c : columns_.entrySet())
{
sc.addColumn(c.getValue().deepCopy());
}
return sc;
}
public IColumn reconcile(IColumn c)
{

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

@ -125,10 +125,9 @@ public abstract class AbstractReplicationStrategy
return WriteResponseHandler.create(writeEndpoints, hintedEndpoints, consistencyLevel, table);
}
// instance method so test subclasses can override it
int getReplicationFactor()
public int getReplicationFactor()
{
return DatabaseDescriptor.getReplicationFactor(table);
return DatabaseDescriptor.getTableDefinition(table).replicationFactor;
}
/**

14
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -29,9 +29,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ServerSocketChannel;
import java.security.MessageDigest;
import java.util.EnumMap;
import java.util.HashMap;
import java.util.Map;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@ -226,7 +224,7 @@ public class MessagingService implements MessagingServiceMBean
* @return an reference to an IAsyncResult which can be queried for the
* response
*/
public String sendRR(Message message, InetAddress[] to, IAsyncCallback cb)
public String sendRR(Message message, Collection<InetAddress> to, IAsyncCallback cb)
{
String messageId = message.getMessageId();
addCallback(cb, messageId);
@ -273,18 +271,16 @@ public class MessagingService implements MessagingServiceMBean
* suggest that a timeout occured to the invoker of the send().
* @return an reference to message id used to match with the result
*/
public String sendRR(Message[] messages, InetAddress[] to, IAsyncCallback cb)
public String sendRR(Message[] messages, List<InetAddress> to, IAsyncCallback cb)
{
if ( messages.length != to.length )
{
if (messages.length != to.size())
throw new IllegalArgumentException("Number of messages and the number of endpoints need to be same.");
}
String groupId = GuidGenerator.guid();
addCallback(cb, groupId);
for ( int i = 0; i < messages.length; ++i )
{
messages[i].setMessageId(groupId);
sendOneWay(messages[i], to[i]);
sendOneWay(messages[i], to.get(i));
}
return groupId;
}

7
src/java/org/apache/cassandra/service/ConsistencyChecker.java
View File

@ -156,7 +156,6 @@ class ConsistencyChecker implements Runnable
static class DataRepairHandler implements IAsyncCallback
{
private final Collection<Message> responses_ = new LinkedBlockingQueue<Message>();
private final ReadResponseResolver readResponseResolver_;
private final int majority_;
@ -167,7 +166,6 @@ class ConsistencyChecker implements Runnable
// wrap localRow in a response Message so it doesn't need to be special-cased in the resolver
ReadResponse readResponse = new ReadResponse(localRow);
Message fakeMessage = new Message(FBUtilities.getLocalAddress(), StorageService.Verb.INTERNAL_RESPONSE, ArrayUtils.EMPTY_BYTE_ARRAY);
responses_.add(fakeMessage);
readResponseResolver_.injectPreProcessed(fakeMessage, readResponse);
}
@ -176,15 +174,14 @@ class ConsistencyChecker implements Runnable
{
if (logger_.isDebugEnabled())
logger_.debug("Received response in DataRepairHandler : " + message.toString());
responses_.add(message);
readResponseResolver_.preprocess(message);
if (responses_.size() == majority_)
if (readResponseResolver_.getMessageCount() == majority_)
{
Runnable runnable = new WrappedRunnable()
{
public void runMayThrow() throws IOException, DigestMismatchException
{
readResponseResolver_.resolve(responses_);
readResponseResolver_.resolve();
}
};
// give remaining replicas until timeout to reply and get added to responses_

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

@ -21,6 +21,9 @@ package org.apache.cassandra.service;
*/
import java.net.InetAddress;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -29,6 +32,7 @@ import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.FBUtilities;
/**
@ -49,14 +53,14 @@ public class DatacenterQuorumResponseHandler<T> extends QuorumResponseHandler<T>
@Override
public void response(Message message)
{
responses.add(message); // we'll go ahead and resolve a reply from anyone, even if it's not from this dc
resolver.preprocess(message);
int n;
n = localdc.equals(snitch.getDatacenter(message.getFrom()))
? localResponses.decrementAndGet()
: localResponses.get();
if (n == 0 && responseResolver.isDataPresent(responses))
if (n == 0 && resolver.isDataPresent())
{
condition.signal();
}
@ -68,4 +72,18 @@ public class DatacenterQuorumResponseHandler<T> extends QuorumResponseHandler<T>
NetworkTopologyStrategy stategy = (NetworkTopologyStrategy) Table.open(table).getReplicationStrategy();
return (stategy.getReplicationFactor(localdc) / 2) + 1;
}
@Override
public void assureSufficientLiveNodes(Collection<InetAddress> endpoints) throws UnavailableException
{
int localEndpoints = 0;
for (InetAddress endpoint : endpoints)
{
if (localdc.equals(snitch.getDatacenter(endpoint)))
localEndpoints++;
}
if(localEndpoints < blockfor)
throw new UnavailableException();
}
}

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

@ -34,8 +34,10 @@ public interface IResponseResolver<T> {
* repairs . Hence you need to derive a response resolver based on your
* needs from this interface.
*/
public T resolve(Collection<Message> responses) throws DigestMismatchException, IOException;
public boolean isDataPresent(Collection<Message> responses);
public T resolve() throws DigestMismatchException, IOException;
public boolean isDataPresent();
public void preprocess(Message message);
public Iterable<Message> getMessages();
public int getMessageCount();
}

48
src/java/org/apache/cassandra/service/QuorumResponseHandler.java
View File

@ -18,19 +18,20 @@
package org.apache.cassandra.service;
import java.io.IOException;
import java.net.InetAddress;
import java.util.Collection;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.io.IOException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.net.IAsyncCallback;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.SimpleCondition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -38,19 +39,20 @@ public class QuorumResponseHandler<T> implements IAsyncCallback
{
protected static final Logger logger = LoggerFactory.getLogger( QuorumResponseHandler.class );
protected final SimpleCondition condition = new SimpleCondition();
protected final Collection<Message> responses = new LinkedBlockingQueue<Message>();;
protected IResponseResolver<T> responseResolver;
protected final IResponseResolver<T> resolver;
private final long startTime;
protected int blockfor;
protected final int blockfor;
/**
* Constructor when response count has to be calculated and blocked for.
*/
public QuorumResponseHandler(IResponseResolver<T> responseResolver, ConsistencyLevel consistencyLevel, String table)
public QuorumResponseHandler(IResponseResolver<T> resolver, ConsistencyLevel consistencyLevel, String table)
{
this.blockfor = determineBlockFor(consistencyLevel, table);
this.responseResolver = responseResolver;
this.resolver = resolver;
this.startTime = System.currentTimeMillis();
logger.debug("QuorumResponseHandler blocking for {} responses", blockfor);
}
public T get() throws TimeoutException, DigestMismatchException, IOException
@ -71,35 +73,31 @@ public class QuorumResponseHandler<T> implements IAsyncCallback
if (!success)
{
StringBuilder sb = new StringBuilder("");
for (Message message : responses)
for (Message message : resolver.getMessages())
{
sb.append(message.getFrom());
}
throw new TimeoutException("Operation timed out - received only " + responses.size() + " responses from " + sb.toString() + " .");
throw new TimeoutException("Operation timed out - received only " + resolver.getMessageCount() + " responses from " + sb.toString() + " .");
}
}
finally
{
for (Message response : responses)
for (Message response : resolver.getMessages())
{
MessagingService.removeRegisteredCallback(response.getMessageId());
}
}
return responseResolver.resolve(responses);
return resolver.resolve();
}
public void response(Message message)
{
responses.add(message);
responseResolver.preprocess(message);
if (responses.size() < blockfor) {
resolver.preprocess(message);
if (resolver.getMessageCount() < blockfor)
return;
}
if (responseResolver.isDataPresent(responses))
{
if (resolver.isDataPresent())
condition.signal();
}
}
public int determineBlockFor(ConsistencyLevel consistencyLevel, String table)
@ -110,11 +108,17 @@ public class QuorumResponseHandler<T> implements IAsyncCallback
case ANY:
return 1;
case QUORUM:
return (DatabaseDescriptor.getReplicationFactor(table) / 2) + 1;
return (Table.open(table).getReplicationStrategy().getReplicationFactor() / 2) + 1;
case ALL:
return DatabaseDescriptor.getReplicationFactor(table);
return Table.open(table).getReplicationStrategy().getReplicationFactor();
default:
throw new UnsupportedOperationException("invalid consistency level: " + table.toString());
throw new UnsupportedOperationException("invalid consistency level: " + consistencyLevel);
}
}
public void assureSufficientLiveNodes(Collection<InetAddress> endpoints) throws UnavailableException
{
if (endpoints.size() < blockfor)
throw new UnavailableException();
}
}

19
src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.service;
import java.io.IOException;
import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -45,6 +46,7 @@ public class RangeSliceResponseResolver implements IResponseResolver<List<Row>>
private static final Logger logger_ = LoggerFactory.getLogger(RangeSliceResponseResolver.class);
private final String table;
private final List<InetAddress> sources;
protected final Collection<Message> responses = new LinkedBlockingQueue<Message>();;
public RangeSliceResponseResolver(String table, List<InetAddress> sources)
{
@ -53,7 +55,7 @@ public class RangeSliceResponseResolver implements IResponseResolver<List<Row>>
this.table = table;
}
public List<Row> resolve(Collection<Message> responses) throws DigestMismatchException, IOException
public List<Row> resolve() throws DigestMismatchException, IOException
{
CollatingIterator collator = new CollatingIterator(new Comparator<Pair<Row,InetAddress>>()
{
@ -110,11 +112,12 @@ public class RangeSliceResponseResolver implements IResponseResolver<List<Row>>
public void preprocess(Message message)
{
responses.add(message);
}
public boolean isDataPresent(Collection<Message> responses)
public boolean isDataPresent()
{
return responses.size() >= sources.size();
return !responses.isEmpty();
}
private static class RowIterator extends AbstractIterator<Pair<Row,InetAddress>>
@ -134,4 +137,14 @@ public class RangeSliceResponseResolver implements IResponseResolver<List<Row>>
return iter.hasNext() ? new Pair<Row, InetAddress>(iter.next(), source) : endOfData();
}
}
public Iterable<Message> getMessages()
{
return responses;
}
public int getMessageCount()
{
return responses.size();
}
}

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

@ -58,14 +58,14 @@ public class ReadResponseResolver implements IResponseResolver<Row>
* repair request should be scheduled.
*
*/
public Row resolve(Collection<Message> responses) throws DigestMismatchException, IOException
public Row resolve() throws DigestMismatchException, IOException
{
if (logger_.isDebugEnabled())
logger_.debug("resolving " + responses.size() + " responses");
logger_.debug("resolving " + results.size() + " responses");
long startTime = System.currentTimeMillis();
List<ColumnFamily> versions = new ArrayList<ColumnFamily>(responses.size());
List<InetAddress> endpoints = new ArrayList<InetAddress>(responses.size());
List<ColumnFamily> versions = new ArrayList<ColumnFamily>();
List<InetAddress> endpoints = new ArrayList<InetAddress>();
DecoratedKey key = null;
ByteBuffer digest = FBUtilities.EMPTY_BYTE_BUFFER;
boolean isDigestQuery = false;
@ -76,11 +76,10 @@ public class ReadResponseResolver implements IResponseResolver<Row>
* query exists then we need to compare the digest with
* the digest of the data that is received.
*/
for (Message message : responses)
{
ReadResponse result = results.get(message);
if (result == null)
continue; // arrived after quorum already achieved
for (Map.Entry<Message, ReadResponse> entry : results.entrySet())
{
ReadResponse result = entry.getValue();
Message message = entry.getKey();
if (result.isDigestQuery())
{
digest = result.digest();
@ -187,6 +186,8 @@ public class ReadResponseResolver implements IResponseResolver<Row>
try
{
ReadResponse result = ReadResponse.serializer().deserialize(new DataInputStream(bufIn));
if (logger_.isDebugEnabled())
logger_.debug("Preprocessed {} response", result.isDigestQuery() ? "digest" : "data");
results.put(message, result);
}
catch (IOException e)
@ -201,16 +202,23 @@ public class ReadResponseResolver implements IResponseResolver<Row>
results.put(message, result);
}
public boolean isDataPresent(Collection<Message> responses)
public boolean isDataPresent()
{
for (Message message : responses)
for (ReadResponse result : results.values())
{
ReadResponse result = results.get(message);
if (result == null)
continue; // arrived concurrently
if (!result.isDigestQuery())
return true;
}
return false;
}
public Iterable<Message> getMessages()
{
return results.keySet();
}
public int getMessageCount()
{
return results.size();
}
}

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

@ -201,7 +201,7 @@ public class StorageProxy implements StorageProxyMBean
{
public void runMayThrow() throws IOException
{
rm.apply();
rm.deepCopy().apply();
responseHandler.response(null);
}
};
@ -315,7 +315,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<List<InetAddress>> commandEndpoints = new ArrayList<List<InetAddress>>();
List<Row> rows = new ArrayList<Row>();
// send out read requests
@ -328,25 +328,25 @@ public class StorageProxy implements StorageProxyMBean
Message messageDigestOnly = readMessageDigestOnly.makeReadMessage();
InetAddress dataPoint = StorageService.instance.findSuitableEndpoint(command.table, command.key);
List<InetAddress> endpointList = StorageService.instance.getLiveNaturalEndpoints(command.table, command.key);
List<InetAddress> endpoints = StorageService.instance.getLiveNaturalEndpoints(command.table, command.key);
InetAddress[] endpoints = new InetAddress[endpointList.size()];
Message messages[] = new Message[endpointList.size()];
AbstractReplicationStrategy rs = Table.open(command.table).getReplicationStrategy();
QuorumResponseHandler<Row> handler = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), consistency_level);
handler.assureSufficientLiveNodes(endpoints);
Message messages[] = new Message[endpoints.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.
int n = 0;
for (InetAddress endpoint : endpointList)
for (InetAddress endpoint : endpoints)
{
Message m = endpoint.equals(dataPoint) ? message : messageDigestOnly;
endpoints[n] = endpoint;
messages[n++] = m;
if (logger.isDebugEnabled())
logger.debug("strongread reading " + (m == message ? "data" : "digest") + " for " + command + " from " + m.getMessageId() + "@" + endpoint);
}
AbstractReplicationStrategy rs = Table.open(command.table).getReplicationStrategy();
QuorumResponseHandler<Row> quorumResponseHandler = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), consistency_level);
MessagingService.instance.sendRR(messages, endpoints, quorumResponseHandler);
quorumResponseHandlers.add(quorumResponseHandler);
MessagingService.instance.sendRR(messages, endpoints, handler);
quorumResponseHandlers.add(handler);
commandEndpoints.add(endpoints);
}
@ -370,14 +370,14 @@ public class StorageProxy implements StorageProxyMBean
catch (DigestMismatchException ex)
{
AbstractReplicationStrategy rs = Table.open(command.table).getReplicationStrategy();
QuorumResponseHandler<Row> qrhRepair = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), ConsistencyLevel.QUORUM);
QuorumResponseHandler<Row> handler = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), consistency_level);
if (logger.isDebugEnabled())
logger.debug("Digest mismatch:", ex);
Message messageRepair = command.makeReadMessage();
MessagingService.instance.sendRR(messageRepair, commandEndpoints.get(i), qrhRepair);
MessagingService.instance.sendRR(messageRepair, commandEndpoints.get(i), handler);
if (repairResponseHandlers == null)
repairResponseHandlers = new ArrayList<QuorumResponseHandler<Row>>();
repairResponseHandlers.add(qrhRepair);
repairResponseHandlers.add(handler);
}
}
@ -499,7 +499,7 @@ public class StorageProxy implements StorageProxyMBean
final Message msg = new Message(FBUtilities.getLocalAddress(), StorageService.Verb.SCHEMA_CHECK, ArrayUtils.EMPTY_BYTE_ARRAY);
final CountDownLatch latch = new CountDownLatch(liveHosts.size());
// an empty message acts as a request to the SchemaCheckVerbHandler.
MessagingService.instance.sendRR(msg, liveHosts.toArray(new InetAddress[]{}), new IAsyncCallback()
MessagingService.instance.sendRR(msg, liveHosts, new IAsyncCallback()
{
public void response(Message msg)
{
@ -774,7 +774,7 @@ public class StorageProxy implements StorageProxyMBean
logger.debug("Starting to send truncate messages to hosts {}", allEndpoints);
Truncation truncation = new Truncation(keyspace, cfname);
Message message = truncation.makeTruncationMessage();
MessagingService.instance.sendRR(message, allEndpoints.toArray(new InetAddress[]{}), responseHandler);
MessagingService.instance.sendRR(message, allEndpoints, responseHandler);
// Wait for all
logger.debug("Sent all truncate messages, now waiting for {} responses", blockFor);

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

@ -1738,7 +1738,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
for (String table : DatabaseDescriptor.getNonSystemTables())
{
// if the replication factor is 1 the data is lost so we shouldn't wait for confirmation
if (DatabaseDescriptor.getReplicationFactor(table) == 1)
if (Table.open(table).getReplicationStrategy().getReplicationFactor() == 1)
continue;
// get all ranges that change ownership (that is, a node needs

5
src/java/org/apache/cassandra/service/WriteResponseHandler.java
View File

@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.Multimap;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.UnavailableException;
@ -93,9 +94,9 @@ public class WriteResponseHandler extends AbstractWriteResponseHandler
}
// at most one node per range can bootstrap at a time, and these will be added to the write until
// bootstrap finishes (at which point we no longer need to write to the old ones).
assert 1 <= blockFor && blockFor <= 2 * DatabaseDescriptor.getReplicationFactor(table)
assert 1 <= blockFor && blockFor <= 2 * Table.open(table).getReplicationStrategy().getReplicationFactor()
: String.format("invalid response count %d for replication factor %d",
blockFor, DatabaseDescriptor.getReplicationFactor(table));
blockFor, Table.open(table).getReplicationStrategy().getReplicationFactor());
return blockFor;
}

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

@ -138,6 +138,7 @@ public class CassandraServer implements Cassandra.Iface
}
catch (TimeoutException e)
{
logger.debug("... timed out");
throw new TimedOutException();
}
catch (IOException e)
@ -260,8 +261,7 @@ public class CassandraServer implements Cassandra.Iface
public List<ColumnOrSuperColumn> get_slice(ByteBuffer key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("get_slice");
logger.debug("get_slice");
state().hasColumnFamilyAccess(column_parent.column_family, Permission.READ);
return multigetSliceInternal(state().getKeyspace(), Collections.singletonList(key), column_parent, predicate, consistency_level).get(key);
@ -270,8 +270,7 @@ public class CassandraServer implements Cassandra.Iface
public Map<ByteBuffer, List<ColumnOrSuperColumn>> multiget_slice(List<ByteBuffer> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("multiget_slice");
logger.debug("multiget_slice");
state().hasColumnFamilyAccess(column_parent.column_family, Permission.READ);
@ -309,8 +308,7 @@ public class CassandraServer implements Cassandra.Iface
public ColumnOrSuperColumn get(ByteBuffer key, ColumnPath column_path, ConsistencyLevel consistency_level)
throws InvalidRequestException, NotFoundException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("get");
logger.debug("get");
state().hasColumnFamilyAccess(column_path.column_family, Permission.READ);
String keyspace = state().getKeyspace();
@ -338,8 +336,7 @@ public class CassandraServer implements Cassandra.Iface
public int get_count(ByteBuffer key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("get_count");
logger.debug("get_count");
state().hasColumnFamilyAccess(column_parent.column_family, Permission.READ);
@ -349,8 +346,7 @@ public class CassandraServer implements Cassandra.Iface
public Map<ByteBuffer, Integer> multiget_count(List<ByteBuffer> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("multiget_count");
logger.debug("multiget_count");
state().hasColumnFamilyAccess(column_parent.column_family, Permission.READ);
String keyspace = state().getKeyspace();
@ -367,8 +363,7 @@ public class CassandraServer implements Cassandra.Iface
public void insert(ByteBuffer key, ColumnParent column_parent, Column column, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("insert");
logger.debug("insert");
state().hasColumnFamilyAccess(column_parent.column_family, Permission.WRITE);
@ -391,8 +386,7 @@ public class CassandraServer implements Cassandra.Iface
public void batch_mutate(Map<ByteBuffer,Map<String,List<Mutation>>> mutation_map, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("batch_mutate");
logger.debug("batch_mutate");
List<String> cfamsSeen = new ArrayList<String>();
@ -428,8 +422,7 @@ public class CassandraServer implements Cassandra.Iface
public void remove(ByteBuffer key, ColumnPath column_path, long timestamp, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("remove");
logger.debug("remove");
state().hasColumnFamilyAccess(column_path.column_family, Permission.WRITE);
@ -450,11 +443,12 @@ public class CassandraServer implements Cassandra.Iface
try
{
StorageProxy.mutate(mutations, consistency_level);
StorageProxy.mutate(mutations, consistency_level);
}
catch (TimeoutException e)
{
throw new TimedOutException();
logger.debug("... timed out");
throw new TimedOutException();
}
}
finally
@ -482,8 +476,7 @@ public class CassandraServer implements Cassandra.Iface
public List<KeySlice> get_range_slices(ColumnParent column_parent, SlicePredicate predicate, KeyRange range, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("range_slice");
logger.debug("range_slice");
String keyspace = state().getKeyspace();
state().hasColumnFamilyAccess(column_parent.column_family, Permission.READ);
@ -521,6 +514,7 @@ public class CassandraServer implements Cassandra.Iface
}
catch (TimeoutException e)
{
logger.debug("... timed out");
throw new TimedOutException();
}
catch (IOException e)
@ -546,8 +540,7 @@ public class CassandraServer implements Cassandra.Iface
public List<KeySlice> get_indexed_slices(ColumnParent column_parent, IndexClause index_clause, SlicePredicate column_predicate, ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException, TException
{
if (logger.isDebugEnabled())
logger.debug("scan");
logger.debug("scan");
state().hasColumnFamilyAccess(column_parent.column_family, Permission.READ);
String keyspace = state().getKeyspace();
@ -566,6 +559,7 @@ public class CassandraServer implements Cassandra.Iface
}
catch (TimeoutException e)
{
logger.debug("... timed out");
throw new TimedOutException();
}
return thriftifyKeySlices(rows, column_parent, column_predicate);
@ -703,6 +697,7 @@ public class CassandraServer implements Cassandra.Iface
public String system_add_column_family(CfDef cf_def) throws InvalidRequestException, TException
{
logger.debug("add_column_family");
state().hasColumnFamilyListAccess(Permission.WRITE);
ThriftValidation.validateCfDef(cf_def);
try
@ -726,6 +721,7 @@ public class CassandraServer implements Cassandra.Iface
public String system_drop_column_family(String column_family) throws InvalidRequestException, TException
{
logger.debug("drop_column_family");
state().hasColumnFamilyListAccess(Permission.WRITE);
try
@ -749,6 +745,7 @@ public class CassandraServer implements Cassandra.Iface
public String system_add_keyspace(KsDef ks_def) throws InvalidRequestException, TException
{
logger.debug("add_keyspace");
state().hasKeyspaceListAccess(Permission.WRITE);
// generate a meaningful error if the user setup keyspace and/or column definition incorrectly
@ -765,6 +762,7 @@ public class CassandraServer implements Cassandra.Iface
Collection<CFMetaData> cfDefs = new ArrayList<CFMetaData>(ks_def.cf_defs.size());
for (CfDef cfDef : ks_def.cf_defs)
{
ThriftValidation.validateCfDef(cfDef);
cfDefs.add(convertToCFMetaData(cfDef));
}
@ -792,6 +790,7 @@ public class CassandraServer implements Cassandra.Iface
public String system_drop_keyspace(String keyspace) throws InvalidRequestException, TException
{
logger.debug("drop_keyspace");
state().hasKeyspaceListAccess(Permission.WRITE);
try
@ -816,6 +815,7 @@ public class CassandraServer implements Cassandra.Iface
/** update an existing keyspace, but do not allow column family modifications. */
public String system_update_keyspace(KsDef ks_def) throws InvalidRequestException, TException
{
logger.debug("update_keyspace");
state().hasKeyspaceListAccess(Permission.WRITE);
ThriftValidation.validateTable(ks_def.name);
@ -848,6 +848,7 @@ public class CassandraServer implements Cassandra.Iface
public String system_update_column_family(CfDef cf_def) throws InvalidRequestException, TException
{
logger.debug("update_column_family");
state().hasColumnFamilyListAccess(Permission.WRITE);
if (cf_def.keyspace == null || cf_def.name == null)

14
src/java/org/apache/cassandra/thrift/ThriftValidation.java
View File

@ -383,14 +383,20 @@ public class ThriftValidation
{
try
{
ColumnFamilyType cfType = ColumnFamilyType.create(cf_def.column_type);
if (cfType == null)
throw new InvalidRequestException("invalid column type " + cf_def.column_type);
DatabaseDescriptor.getComparator(cf_def.comparator_type);
DatabaseDescriptor.getComparator(cf_def.subcomparator_type);
DatabaseDescriptor.getComparator(cf_def.default_validation_class);
if (cfType != ColumnFamilyType.Super && cf_def.subcomparator_type != null)
throw new InvalidRequestException("subcomparator_type is invalid for standard columns");
if (cf_def.column_metadata == null)
return;
AbstractType comparator = cf_def.subcomparator_type == null
AbstractType comparator = cfType == ColumnFamilyType.Standard
? DatabaseDescriptor.getComparator(cf_def.comparator_type)
: DatabaseDescriptor.getComparator(cf_def.subcomparator_type);
for (ColumnDef c : cf_def.column_metadata)
@ -406,6 +412,12 @@ public class ThriftValidation
throw new InvalidRequestException(String.format("Column name %s is not valid for comparator %s",
FBUtilities.bytesToHex(c.name), cf_def.comparator_type));
}
if ((c.index_name != null) && (c.index_type == null))
throw new ConfigurationException("index_name cannot be set without index_type");
if (cfType == ColumnFamilyType.Super && c.index_type != null)
throw new InvalidRequestException("Secondary indexes are not supported on supercolumns");
}
}
catch (ConfigurationException e)

14
src/java/org/apache/cassandra/utils/ByteBufferUtil.java
View File

@ -60,8 +60,8 @@ import java.nio.charset.Charset;
* }
*
*/
public class ByteBufferUtil {
public class ByteBufferUtil
{
public static int compareUnsigned(ByteBuffer o1, ByteBuffer o2)
{
return FBUtilities.compareUnsigned(o1.array(), o2.array(), o1.arrayOffset()+o1.position(), o2.arrayOffset()+o2.position(), o1.limit()+o1.arrayOffset(), o2.limit()+o2.arrayOffset());
@ -98,4 +98,14 @@ public class ByteBufferUtil {
throw new RuntimeException(e);
}
}
public static ByteBuffer clone(ByteBuffer o)
{
ByteBuffer clone = ByteBuffer.allocate(o.remaining());
o.mark();
clone.put(o);
o.reset();
clone.flip();
return clone;
}
}

4
test/unit/org/apache/cassandra/cli/CliTest.java
View File

@ -36,11 +36,13 @@ public class CliTest extends CleanupHelper
// please add new statements here so they could be auto-runned by this test.
private String[] statements = {
"use TestKeySpace;",
"create column family CF1 with comparator=UTF8Type and column_metadata=[{ column_name:world, validation_class:IntegerType, index_type:0, index_name:IdxName }, { column_name:world2, validation_class:LongType, index_type:0, index_name:LongIdxName}];",
"create column family CF1 with comparator=UTF8Type and column_metadata=[{ column_name:world, validation_class:IntegerType, index_type:0, index_name:IdxName }, { column_name:world2, validation_class:LongType, index_type:KEYS, index_name:LongIdxName}];",
"set CF1[hello][world] = 123848374878933948398384;",
"get CF1[hello][world];",
"set CF1[hello][world2] = 15;",
"get CF1 where world2 = long(15);",
"get cF1 where world2 = long(15);",
"get Cf1 where world2 = long(15);",
"set CF1['hello'][time_spent_uuid] = timeuuid(a8098c1a-f86e-11da-bd1a-00112444be1e);",
"create column family CF2 with comparator=IntegerType;",
"set CF2['key'][98349387493847748398334] = 'some text';",

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

@ -33,6 +33,7 @@ import org.junit.Test;
import com.google.common.collect.Multimap;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.IFailureDetectionEventListener;
import org.apache.cassandra.gms.IFailureDetector;
@ -146,7 +147,7 @@ public class BootStrapperTest extends CleanupHelper
final int[] clusterSizes = new int[] { 1, 3, 5, 10, 100};
for (String table : DatabaseDescriptor.getNonSystemTables())
{
int replicationFactor = DatabaseDescriptor.getReplicationFactor(table);
int replicationFactor = Table.open(table).getReplicationStrategy().getReplicationFactor();
for (int clusterSize : clusterSizes)
if (clusterSize >= replicationFactor)
testSourceTargetComputation(table, clusterSize, replicationFactor);

22
test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java
View File

@ -30,11 +30,12 @@ import org.junit.Before;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.dht.BigIntegerToken;
import org.apache.cassandra.dht.Token;
public class OldNetworkTopologyStrategyTest
public class OldNetworkTopologyStrategyTest extends SchemaLoader
{
private List<Token> endpointTokens;
private List<Token> keyTokens;
@ -71,7 +72,7 @@ public class OldNetworkTopologyStrategyTest
expectedResults.put("25", buildResult("254.0.0.4", "254.0.0.1", "254.0.0.2"));
expectedResults.put("35", buildResult("254.0.0.1", "254.0.0.2", "254.0.0.3"));
runTestForReplicatedTables(strategy);
testGetEndpoints(strategy, keyTokens.toArray(new Token[0]));
}
/**
@ -96,7 +97,7 @@ public class OldNetworkTopologyStrategyTest
expectedResults.put("25", buildResult("254.0.0.4", "254.1.0.3", "254.0.0.1"));
expectedResults.put("35", buildResult("254.0.0.1", "254.1.0.3", "254.0.0.2"));
runTestForReplicatedTables(strategy);
testGetEndpoints(strategy, keyTokens.toArray(new Token[0]));
}
/**
@ -122,16 +123,7 @@ public class OldNetworkTopologyStrategyTest
expectedResults.put("25", buildResult("254.1.0.4", "254.0.0.1", "254.0.0.2"));
expectedResults.put("35", buildResult("254.0.0.1", "254.0.1.3", "254.1.0.4"));
runTestForReplicatedTables(strategy);
}
private void runTestForReplicatedTables(AbstractReplicationStrategy strategy) throws UnknownHostException
{
for (String table : DatabaseDescriptor.getNonSystemTables())
{
if (DatabaseDescriptor.getReplicationFactor(table) == 3)
testGetEndpoints(strategy, keyTokens.toArray(new Token[0]), table);
}
testGetEndpoints(strategy, keyTokens.toArray(new Token[0]));
}
private ArrayList<InetAddress> buildResult(String... addresses) throws UnknownHostException
@ -156,7 +148,7 @@ public class OldNetworkTopologyStrategyTest
tmd.updateNormalToken(endpointToken, ep);
}
private void testGetEndpoints(AbstractReplicationStrategy strategy, Token[] keyTokens, String table) throws UnknownHostException
private void testGetEndpoints(AbstractReplicationStrategy strategy, Token[] keyTokens) throws UnknownHostException
{
for (Token keyToken : keyTokens)
{

4
test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java
View File

@ -94,7 +94,7 @@ public class SimpleStrategyTest extends CleanupHelper
for (int i = 0; i < keyTokens.length; i++)
{
List<InetAddress> endpoints = strategy.getNaturalEndpoints(keyTokens[i]);
assertEquals(DatabaseDescriptor.getReplicationFactor(table), endpoints.size());
assertEquals(strategy.getReplicationFactor(), endpoints.size());
List<InetAddress> correctEndpoints = new ArrayList<InetAddress>();
for (int j = 0; j < endpoints.size(); j++)
correctEndpoints.add(hosts.get((i + j + 1) % hosts.size()));
@ -140,7 +140,7 @@ public class SimpleStrategyTest extends CleanupHelper
StorageService.calculatePendingRanges(strategy, table);
int replicationFactor = DatabaseDescriptor.getReplicationFactor(table);
int replicationFactor = strategy.getReplicationFactor();
for (int i = 0; i < keyTokens.length; i++)
{

4
test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java
View File

@ -171,7 +171,7 @@ public class AntiEntropyServiceTest extends CleanupHelper
public void testGetNeighborsPlusOne() throws Throwable
{
// generate rf+1 nodes, and ensure that all nodes are returned
Set<InetAddress> expected = addTokens(1 + DatabaseDescriptor.getReplicationFactor(tablename));
Set<InetAddress> expected = addTokens(1 + Table.open(tablename).getReplicationStrategy().getReplicationFactor());
expected.remove(FBUtilities.getLocalAddress());
assertEquals(expected, AntiEntropyService.getNeighbors(tablename));
}
@ -182,7 +182,7 @@ public class AntiEntropyServiceTest extends CleanupHelper
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
// generate rf*2 nodes, and ensure that only neighbors specified by the ARS are returned
addTokens(2 * DatabaseDescriptor.getReplicationFactor(tablename));
addTokens(2 * Table.open(tablename).getReplicationStrategy().getReplicationFactor());
AbstractReplicationStrategy ars = Table.open(tablename).getReplicationStrategy();
Set<InetAddress> expected = new HashSet<InetAddress>();
for (Range replicaRange : ars.getAddressRanges().get(FBUtilities.getLocalAddress()))

166
test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java Normal file
View File

@ -0,0 +1,166 @@
package org.apache.cassandra.service;
/*
*
* 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.
*
*/
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import com.google.common.collect.HashMultimap;
import org.junit.Test;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Row;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.SimpleSnitch;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.UnavailableException;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class ConsistencyLevelTest extends CleanupHelper
{
@Test
public void testReadWriteConsistencyChecks() throws Exception
{
StorageService ss = StorageService.instance;
final int RING_SIZE = 3;
TokenMetadata tmd = ss.getTokenMetadata();
tmd.clearUnsafe();
IPartitioner partitioner = new RandomPartitioner();
ss.setPartitionerUnsafe(partitioner);
ArrayList<Token> endpointTokens = new ArrayList<Token>();
ArrayList<Token> keyTokens = new ArrayList<Token>();
List<InetAddress> hosts = new ArrayList<InetAddress>();
Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, RING_SIZE);
HashMultimap<InetAddress, InetAddress> hintedNodes = HashMultimap.create();
AbstractReplicationStrategy strategy;
for (String table : DatabaseDescriptor.getNonSystemTables())
{
strategy = getStrategy(table, tmd);
StorageService.calculatePendingRanges(strategy, table);
int replicationFactor = strategy.getReplicationFactor();
if (replicationFactor < 2)
continue;
for (ConsistencyLevel c : ConsistencyLevel.values())
{
if (c == ConsistencyLevel.EACH_QUORUM || c == ConsistencyLevel.LOCAL_QUORUM)
continue;
for (int i = 0; i < replicationFactor; i++)
{
hintedNodes.clear();
for (int j = 0; j < i; j++)
{
hintedNodes.put(hosts.get(j), hosts.get(j));
}
IWriteResponseHandler writeHandler = strategy.getWriteResponseHandler(hosts, hintedNodes, c);
QuorumResponseHandler<Row> readHandler = strategy.getQuorumResponseHandler(new ReadResponseResolver(table), c);
boolean isWriteUnavailable = false;
boolean isReadUnavailable = false;
try
{
writeHandler.assureSufficientLiveNodes();
}
catch (UnavailableException e)
{
isWriteUnavailable = true;
}
try
{
readHandler.assureSufficientLiveNodes(hintedNodes.asMap().keySet());
}
catch (UnavailableException e)
{
isReadUnavailable = true;
}
//these should always match (in this kind of test)
assertTrue(isWriteUnavailable == isReadUnavailable);
switch (c)
{
case ALL:
if (isWriteUnavailable)
assertTrue(hintedNodes.size() < replicationFactor);
else
assertTrue(hintedNodes.size() >= replicationFactor);
break;
case ONE:
case ANY:
if (isWriteUnavailable)
assertTrue(hintedNodes.size() == 0);
else
assertTrue(hintedNodes.size() > 0);
break;
case QUORUM:
if (isWriteUnavailable)
assertTrue(hintedNodes.size() < (replicationFactor / 2 + 1));
else
assertTrue(hintedNodes.size() >= (replicationFactor / 2 + 1));
break;
default:
fail("Unhandled CL: " + c);
}
}
}
return;
}
fail("Test requires at least one table with RF > 1");
}
private AbstractReplicationStrategy getStrategy(String table, TokenMetadata tmd) throws ConfigurationException
{
return AbstractReplicationStrategy.createReplicationStrategy(table,
"org.apache.cassandra.locator.SimpleStrategy",
tmd,
new SimpleSnitch(),
null);
}
}

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

@ -92,7 +92,7 @@ public class MoveTest extends CleanupHelper
strategy = getStrategy(table, tmd);
for (Token token : keyTokens)
{
int replicationFactor = DatabaseDescriptor.getReplicationFactor(table);
int replicationFactor = strategy.getReplicationFactor();
HashSet<InetAddress> actual = new HashSet<InetAddress>(tmd.getWriteEndpoints(token, table, strategy.calculateNaturalEndpoints(token, tmd)));
HashSet<InetAddress> expected = new HashSet<InetAddress>();
@ -217,7 +217,7 @@ public class MoveTest extends CleanupHelper
}
// just to be sure that things still work according to the old tests, run them:
if (DatabaseDescriptor.getReplicationFactor(table) != 3)
if (strategy.getReplicationFactor() != 3)
continue;
// tokens 5, 15 and 25 should go three nodes
for (int i=0; i<3; ++i)
@ -334,7 +334,7 @@ public class MoveTest extends CleanupHelper
assertTrue(expectedEndpoints.get(table).get(keyTokens.get(i)).containsAll(endpoints));
}
if (DatabaseDescriptor.getReplicationFactor(table) != 3)
if (strategy.getReplicationFactor() != 3)
continue;
// leave this stuff in to guarantee the old tests work the way they were supposed to.
// tokens 5, 15 and 25 should go three nodes