diff --git a/CHANGES.txt b/CHANGES.txt index e115d07219..e83069fd40 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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 diff --git a/build.xml b/build.xml index b9f583db37..ff575e6c0a 100644 --- a/build.xml +++ b/build.xml @@ -47,7 +47,7 @@ - + diff --git a/conf/cassandra-env.sh b/conf/cassandra-env.sh index 9a946ad68d..2870fdb34d 100644 --- a/conf/cassandra-env.sh +++ b/conf/cassandra-env.sh @@ -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). diff --git a/contrib/maven/pom.xml b/contrib/maven/pom.xml deleted file mode 100644 index 5d98eb76a8..0000000000 --- a/contrib/maven/pom.xml +++ /dev/null @@ -1,551 +0,0 @@ - - - - - org.apache - apache - 6 - - 4.0.0 - - org.apache.cassandra - cassandra - 0.6-SNAPSHOT - jar - Cassandra - 2009 - - - 2.0.9 - - - http://incubator.apache.org/cassandra - - - - cassandra-user - cassandra-user-subscribe@incubator.apache.org - cassandra-user-unsubscribe@incubator.apache.org - cassandra-user@incubator.apache.org - http://mail-archives.apache.org/mod_mbox/incubator-cassandra-user/ - - - cassandra-dev - cassandra-dev-subscribe@incubator.apache.org - cassandra-dev-unsubscribe@incubator.apache.org - cassandra-dev@incubator.apache.org - http://mail-archives.apache.org/mod_mbox/incubator-cassandra-dev/ - - - cassandra-commits - cassandra-commits-subscribe@incubator.apache.org - cassandra-commits-unsubscribe@incubator.apache.org - cassandra-commits@incubator.apache.org - http://mail-archives.apache.org/mod_mbox/incubator-cassandra-commits/ - - - - - - - - - - - JIRA - https://issues.apache.org/jira/browse/CASSANDRA - - - - - hudson - http://hudson.zones.apache.org/hudson - - - - mail - true - true - false - false -
cassandra-commits@incubator.apache.org
 - - - - - - - cassandra-website - scp://people.apache.org/x1/www/incubator.apache.org/cassandra/maven/${pom.version} - - - - - scm:svn:http://svn.apache.org/repos/asf/incubator/cassandra/trunk - scm:svn:https://svn.apache.org/repos/asf/incubator/cassandra/trunk - http://svn.apache.org/viewvc/incubator/cassandra/trunk/ - - - - - - commons-collections - commons-collections - 3.2.1 - - - commons-cli - commons-cli - 1.1 - - - commons-lang - commons-lang - 2.4 - - - jline - jline - 0.9.94 - - - log4j - log4j - 1.2.15 - - - - javax.jms - jms - - - com.sun.jmx - jmxri - - - com.sun.jdmk - jmxtools - - - - - org.slf4j - slf4j-api - 1.5.8 - - - org.slf4j - slf4j-log4j12 - 1.5.8 - - - org.antlr - antlr-runtime - 3.1.3 - - - com.google.collections - google-collections - 1.0-rc1 - - - - - - high-scale-lib - high-scale-lib - UNKNOWN - system - ${basedir}/lib/high-scale-lib.jar - - - flexjson - flexjson - 1.7 - system - ${basedir}/lib/flexjson-1.7.jar - - - libthrift - libthrift - UNKNOWN - system - ${basedir}/lib/libthrift-r820831.jar - - - jsonsimple - jsonsimple - UNKNOWN - system - ${basedir}/lib/json_simple-1.1.jar - - - com.reardencommerce - clhm - UNKNOWN - system - ${basedir}/lib/clhm-production.jar - - - - - junit - junit - 4.6 - test - - - - - - ${basedir}/src/java - ${basedir}/test/unit - build/classes - - - - ${basedir}/test/conf - - **/* - - - - ${basedir}/test/resources - - *.json - - - - - - - - - org.antlr - antlr3-maven-plugin - 3.1.3-1 - - - process-sources - - antlr - - - ${basedir}/src/java - - - - - - - - org.codehaus.mojo - build-helper-maven-plugin - 1.3 - - - add-source - generate-sources - - add-source - - - - ${basedir}/interface/gen-java - - - - - - - - - org.apache.maven.plugins - maven-compiler-plugin - - 1.6 - 1.6 - true - true - true - true - - - - - - org.apache.maven.plugins - maven-surefire-plugin - - - - storage-config - ${basedir}/test/conf - - - always - - **/TestRingCache.java - - - - - - - org.codehaus.mojo - cobertura-maven-plugin - 2.0 - - - - - - - org.apache.maven.plugins - maven-release-plugin - 2.0-beta-9 - - true - false - clean install - deploy - -Papache-release - - - - org.codehaus.mojo - ianal-maven-plugin - 1.0-alpha-1 - - - org.codehaus.mojo - rat-maven-plugin - 1.0-alpha-3 - - false - - - - org.apache.maven.plugins - maven-enforcer-plugin - - - validate - - enforce - - - - - [2.0.9,) - - - - - - - - org.codehaus.mojo - ianal-maven-plugin - - - - verify-legal-files - - - true - - - - - - - - - - - org.apache.maven.plugins - maven-jxr-plugin - - - org.apache.maven.plugins - maven-surefire-report-plugin - - - org.apache.maven.plugins - maven-pmd-plugin - - - org.codehaus.mojo - taglist-maven-plugin - - - org.apache.maven.plugins - maven-javadoc-plugin - - - http://java.sun.com/j2se/1.6.0/docs/api/ - http://logging.apache.org/log4j/docs/api/ - - - true - 900m - 1.6 - - - - org.codehaus.mojo - cobertura-maven-plugin - 2.2 - - - html - xml - - - - - - - - - - - thrift - - process-sources - - - - maven-antrun-plugin - - - process-sources - - - - - - - - run - - - - - - - - - - - apache-release - - - - - - org.apache.maven.plugins - maven-gpg-plugin - - ${gpg.passphrase} - - - - - sign - - - - - - - - true - org.apache.maven.plugins - maven-deploy-plugin - - true - - - - org.apache.maven.plugins - maven-source-plugin - - - attach-sources - - jar - - - - - - org.apache.maven.plugins - maven-javadoc-plugin - - ${project.build.sourceEncoding} - - - - attach-javadocs - - jar - - - - - - - - org.apache.maven.plugins - maven-assembly-plugin - - - - single - - package - - true - - - source-release - - - - - - - - org.apache.geronimo.genesis - apache-source-release-assembly-descriptor - 2.0 - - - - - - - - - - - diff --git a/debian/changelog b/debian/changelog index 0017e47f0d..765ccda88f 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +cassandra (0.7.0~rc2) unstable; urgency=low + + * Release candidate release. + + -- Eric Evans Mon, 06 Dec 2010 11:19:40 -0600 + cassandra (0.7.0~rc1) unstable; urgency=low * Release candidate release. diff --git a/src/java/org/apache/cassandra/cli/CliClient.java b/src/java/org/apache/cassandra/cli/CliClient.java index 0aafe0c658..192ffa4ee9 100644 --- a/src/java/org/apache/cassandra/cli/CliClient.java +++ b/src/java/org/apache/cassandra/cli/CliClient.java @@ -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; diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index 8f3f496858..59ec9f5a4e 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -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()); diff --git a/src/java/org/apache/cassandra/config/ColumnDefinition.java b/src/java/org/apache/cassandra/config/ColumnDefinition.java index 54f5b09ae6..abb0c7ff75 100644 --- a/src/java/org/apache/cassandra/config/ColumnDefinition.java +++ b/src/java/org/apache/cassandra/config/ColumnDefinition.java @@ -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 cds = new TreeMap(); 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)) diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 5968a62405..7f59afc5c2 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/Column.java b/src/java/org/apache/cassandra/db/Column.java index b61ee42f49..38abb152f5 100644 --- a/src/java/org/apache/cassandra/db/Column.java +++ b/src/java/org/apache/cassandra/db/Column.java @@ -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(); diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyType.java b/src/java/org/apache/cassandra/db/ColumnFamilyType.java index 9f66b41346..d250ecd9b9 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyType.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyType.java @@ -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) { diff --git a/src/java/org/apache/cassandra/db/DeletedColumn.java b/src/java/org/apache/cassandra/db/DeletedColumn.java index 5e7c7ec3f6..7af5af5190 100644 --- a/src/java/org/apache/cassandra/db/DeletedColumn.java +++ b/src/java/org/apache/cassandra/db/DeletedColumn.java @@ -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); + } } diff --git a/src/java/org/apache/cassandra/db/ExpiringColumn.java b/src/java/org/apache/cassandra/db/ExpiringColumn.java index 5496a45e3e..e587618ef6 100644 --- a/src/java/org/apache/cassandra/db/ExpiringColumn.java +++ b/src/java/org/apache/cassandra/db/ExpiringColumn.java @@ -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) { diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index dac514549f..c83bb646c3 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -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(); diff --git a/src/java/org/apache/cassandra/db/IColumn.java b/src/java/org/apache/cassandra/db/IColumn.java index d7f9d3d642..61bb6783c3 100644 --- a/src/java/org/apache/cassandra/db/IColumn.java +++ b/src/java/org/apache/cassandra/db/IColumn.java @@ -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 diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index 426b6ca61b..8365bd556e 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -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 e : modifications_.entrySet()) + { + ColumnFamily cf = e.getValue().cloneMeShallow(); + for (Map.Entry ce : e.getValue().getColumnsMap().entrySet()) + cf.addColumn(ce.getValue().deepCopy()); + rm.modifications_.put(e.getKey(), cf); + } + + return rm; + } } class RowMutationSerializer implements ICompactSerializer diff --git a/src/java/org/apache/cassandra/db/SuperColumn.java b/src/java/org/apache/cassandra/db/SuperColumn.java index c74bf371fd..0d5def4485 100644 --- a/src/java/org/apache/cassandra/db/SuperColumn.java +++ b/src/java/org/apache/cassandra/db/SuperColumn.java @@ -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 c : columns_.entrySet()) + { + sc.addColumn(c.getValue().deepCopy()); + } + + return sc; + } public IColumn reconcile(IColumn c) { diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index 9ec15c114b..1f997cbafc 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -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; } /** diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 5da901c3bb..883eb0a3f7 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -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 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 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; } diff --git a/src/java/org/apache/cassandra/service/ConsistencyChecker.java b/src/java/org/apache/cassandra/service/ConsistencyChecker.java index a30b67ff93..f1bc2f2af7 100644 --- a/src/java/org/apache/cassandra/service/ConsistencyChecker.java +++ b/src/java/org/apache/cassandra/service/ConsistencyChecker.java @@ -156,7 +156,6 @@ class ConsistencyChecker implements Runnable static class DataRepairHandler implements IAsyncCallback { - private final Collection responses_ = new LinkedBlockingQueue(); 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_ diff --git a/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java index 6267df94e9..0180d51dda 100644 --- a/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java @@ -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 extends QuorumResponseHandler @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 extends QuorumResponseHandler NetworkTopologyStrategy stategy = (NetworkTopologyStrategy) Table.open(table).getReplicationStrategy(); return (stategy.getReplicationFactor(localdc) / 2) + 1; } + + @Override + public void assureSufficientLiveNodes(Collection endpoints) throws UnavailableException + { + int localEndpoints = 0; + for (InetAddress endpoint : endpoints) + { + if (localdc.equals(snitch.getDatacenter(endpoint))) + localEndpoints++; + } + + if(localEndpoints < blockfor) + throw new UnavailableException(); + } } diff --git a/src/java/org/apache/cassandra/service/IResponseResolver.java b/src/java/org/apache/cassandra/service/IResponseResolver.java index 0b5b54f64e..ea61705455 100644 --- a/src/java/org/apache/cassandra/service/IResponseResolver.java +++ b/src/java/org/apache/cassandra/service/IResponseResolver.java @@ -34,8 +34,10 @@ public interface IResponseResolver { * repairs . Hence you need to derive a response resolver based on your * needs from this interface. */ - public T resolve(Collection responses) throws DigestMismatchException, IOException; - public boolean isDataPresent(Collection responses); + public T resolve() throws DigestMismatchException, IOException; + public boolean isDataPresent(); public void preprocess(Message message); + public Iterable getMessages(); + public int getMessageCount(); } diff --git a/src/java/org/apache/cassandra/service/QuorumResponseHandler.java b/src/java/org/apache/cassandra/service/QuorumResponseHandler.java index c21aac2e42..9d7c7fdf02 100644 --- a/src/java/org/apache/cassandra/service/QuorumResponseHandler.java +++ b/src/java/org/apache/cassandra/service/QuorumResponseHandler.java @@ -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 implements IAsyncCallback { protected static final Logger logger = LoggerFactory.getLogger( QuorumResponseHandler.class ); protected final SimpleCondition condition = new SimpleCondition(); - protected final Collection responses = new LinkedBlockingQueue();; - protected IResponseResolver responseResolver; + protected final IResponseResolver 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 responseResolver, ConsistencyLevel consistencyLevel, String table) + public QuorumResponseHandler(IResponseResolver 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 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 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 endpoints) throws UnavailableException + { + if (endpoints.size() < blockfor) + throw new UnavailableException(); + } } diff --git a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java index bec0c1296d..8b98019a74 100644 --- a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java +++ b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java @@ -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> private static final Logger logger_ = LoggerFactory.getLogger(RangeSliceResponseResolver.class); private final String table; private final List sources; + protected final Collection responses = new LinkedBlockingQueue();; public RangeSliceResponseResolver(String table, List sources) { @@ -53,7 +55,7 @@ public class RangeSliceResponseResolver implements IResponseResolver> this.table = table; } - public List resolve(Collection responses) throws DigestMismatchException, IOException + public List resolve() throws DigestMismatchException, IOException { CollatingIterator collator = new CollatingIterator(new Comparator>() { @@ -110,11 +112,12 @@ public class RangeSliceResponseResolver implements IResponseResolver> public void preprocess(Message message) { + responses.add(message); } - public boolean isDataPresent(Collection responses) + public boolean isDataPresent() { - return responses.size() >= sources.size(); + return !responses.isEmpty(); } private static class RowIterator extends AbstractIterator> @@ -134,4 +137,14 @@ public class RangeSliceResponseResolver implements IResponseResolver> return iter.hasNext() ? new Pair(iter.next(), source) : endOfData(); } } + + public Iterable getMessages() + { + return responses; + } + + public int getMessageCount() + { + return responses.size(); + } } diff --git a/src/java/org/apache/cassandra/service/ReadResponseResolver.java b/src/java/org/apache/cassandra/service/ReadResponseResolver.java index 4a4d658b64..440f163776 100644 --- a/src/java/org/apache/cassandra/service/ReadResponseResolver.java +++ b/src/java/org/apache/cassandra/service/ReadResponseResolver.java @@ -58,14 +58,14 @@ public class ReadResponseResolver implements IResponseResolver * repair request should be scheduled. * */ - public Row resolve(Collection 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 versions = new ArrayList(responses.size()); - List endpoints = new ArrayList(responses.size()); + List versions = new ArrayList(); + List endpoints = new ArrayList(); DecoratedKey key = null; ByteBuffer digest = FBUtilities.EMPTY_BYTE_BUFFER; boolean isDigestQuery = false; @@ -76,11 +76,10 @@ public class ReadResponseResolver implements IResponseResolver * 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 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 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 results.put(message, result); } - public boolean isDataPresent(Collection 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 getMessages() + { + return results.keySet(); + } + + public int getMessageCount() + { + return results.size(); + } } diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 30104916d8..a1e97d06b8 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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 strongRead(List commands, ConsistencyLevel consistency_level) throws IOException, UnavailableException, TimeoutException { List> quorumResponseHandlers = new ArrayList>(); - List commandEndpoints = new ArrayList(); + List> commandEndpoints = new ArrayList>(); List rows = new ArrayList(); // 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 endpointList = StorageService.instance.getLiveNaturalEndpoints(command.table, command.key); + List 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 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 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 qrhRepair = rs.getQuorumResponseHandler(new ReadResponseResolver(command.table), ConsistencyLevel.QUORUM); + QuorumResponseHandler 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>(); - 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); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 7471aaf55d..2bb83f9838 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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 diff --git a/src/java/org/apache/cassandra/service/WriteResponseHandler.java b/src/java/org/apache/cassandra/service/WriteResponseHandler.java index 3ea15c97f4..95e5e3f362 100644 --- a/src/java/org/apache/cassandra/service/WriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/WriteResponseHandler.java @@ -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; } diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index 028fb6e169..e8d659c465 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -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 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> multiget_slice(List 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 multiget_count(List 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>> mutation_map, ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException { - if (logger.isDebugEnabled()) - logger.debug("batch_mutate"); + logger.debug("batch_mutate"); List cfamsSeen = new ArrayList(); @@ -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 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 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 cfDefs = new ArrayList(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) diff --git a/src/java/org/apache/cassandra/thrift/ThriftValidation.java b/src/java/org/apache/cassandra/thrift/ThriftValidation.java index 57db71be41..062ba6af79 100644 --- a/src/java/org/apache/cassandra/thrift/ThriftValidation.java +++ b/src/java/org/apache/cassandra/thrift/ThriftValidation.java @@ -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) diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java index a96a33001f..80fdd229ef 100644 --- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java +++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java @@ -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; + } } diff --git a/test/unit/org/apache/cassandra/cli/CliTest.java b/test/unit/org/apache/cassandra/cli/CliTest.java index 50d4a01b7d..6b74e78f35 100644 --- a/test/unit/org/apache/cassandra/cli/CliTest.java +++ b/test/unit/org/apache/cassandra/cli/CliTest.java @@ -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';", diff --git a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java index 7fc460203a..d42e87c270 100644 --- a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java +++ b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java b/test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java index c42d7c3e2c..288491d3a8 100644 --- a/test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java +++ b/test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java @@ -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 endpointTokens; private List 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 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) { diff --git a/test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java b/test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java index f53680c783..56d7089ac4 100644 --- a/test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java +++ b/test/unit/org/apache/cassandra/locator/SimpleStrategyTest.java @@ -94,7 +94,7 @@ public class SimpleStrategyTest extends CleanupHelper for (int i = 0; i < keyTokens.length; i++) { List endpoints = strategy.getNaturalEndpoints(keyTokens[i]); - assertEquals(DatabaseDescriptor.getReplicationFactor(table), endpoints.size()); + assertEquals(strategy.getReplicationFactor(), endpoints.size()); List correctEndpoints = new ArrayList(); 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++) { diff --git a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java index ccbdcc0157..7402233469 100644 --- a/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java +++ b/test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java @@ -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 expected = addTokens(1 + DatabaseDescriptor.getReplicationFactor(tablename)); + Set 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 expected = new HashSet(); for (Range replicaRange : ars.getAddressRanges().get(FBUtilities.getLocalAddress())) diff --git a/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java b/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java new file mode 100644 index 0000000000..2ebb5bf395 --- /dev/null +++ b/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java @@ -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 endpointTokens = new ArrayList(); + ArrayList keyTokens = new ArrayList(); + List hosts = new ArrayList(); + + Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, RING_SIZE); + + HashMultimap 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 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); + } + +} diff --git a/test/unit/org/apache/cassandra/service/MoveTest.java b/test/unit/org/apache/cassandra/service/MoveTest.java index ba75a22d93..90b80ef7c3 100644 --- a/test/unit/org/apache/cassandra/service/MoveTest.java +++ b/test/unit/org/apache/cassandra/service/MoveTest.java @@ -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 actual = new HashSet(tmd.getWriteEndpoints(token, table, strategy.calculateNaturalEndpoints(token, tmd))); HashSet expected = new HashSet(); @@ -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