cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r1073768 - in /cassandra/branches/cassandra-0.7: CHANGES.txt src/java/org/apache/cassandra/service/StorageProxy.java src/java/org/apache/cassandra/thrift/CassandraServer.java
Date Wed, 23 Feb 2011 15:09:29 GMT
Author: jbellis
Date: Wed Feb 23 15:09:28 2011
New Revision: 1073768

URL: http://svn.apache.org/viewvc?rev=1073768&view=rev
Log:
add validateSchemaAgreement call + synchronization to schema modification calls
patch by jbellis; reviewed by gdusbabek for CASSANDRA-2222

Modified:
    cassandra/branches/cassandra-0.7/CHANGES.txt
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java

Modified: cassandra/branches/cassandra-0.7/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/CHANGES.txt?rev=1073768&r1=1073767&r2=1073768&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.7/CHANGES.txt Wed Feb 23 15:09:28 2011
@@ -21,6 +21,8 @@
  * fix BRAF performance when seeking to EOF (CASSANDRA-2218)
  * check for memtable flush_after_mins exceeded every 10s (CASSANDRA-2183)
  * fix cache saving on Windows (CASSANDRA-2207)
+ * add validateSchemaAgreement call + synchronization to schema
+   modification operations (CASSANDRA-2222)
 
 
 0.7.2

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java?rev=1073768&r1=1073767&r2=1073768&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
(original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
Wed Feb 23 15:09:28 2011
@@ -616,17 +616,18 @@ public class StorageProxy implements Sto
             }
             hosts.add(host.getHostAddress());
         }
+
+        // we're done: the results map is ready to return to the client.  the rest is just
debug logging:
         if (results.get(UNREACHABLE) != null)
             logger.debug("Hosts not in agreement. Didn't get a response from everybody: "
+ StringUtils.join(results.get(UNREACHABLE), ","));
-        // check for version disagreement. log the hosts that don't agree.
         for (Map.Entry<String, List<String>> entry : results.entrySet())
         {
+            // check for version disagreement. log the hosts that don't agree.
             if (entry.getKey().equals(UNREACHABLE) || entry.getKey().equals(myVersion))
                 continue;
             for (String host : entry.getValue())
                 logger.debug("%s disagrees (%s)", host, entry.getKey());
         }
-        
         if (results.size() == 1)
             logger.debug("Schemas are in agreement.");
         

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java?rev=1073768&r1=1073767&r2=1073768&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java
(original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java
Wed Feb 23 15:09:28 2011
@@ -26,6 +26,8 @@ import java.util.concurrent.ExecutionExc
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeoutException;
 
+import com.google.common.base.Predicates;
+import com.google.common.collect.Maps;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -661,11 +663,13 @@ public class CassandraServer implements 
         }
     }
 
-    public String system_add_column_family(CfDef cf_def) throws InvalidRequestException,
TException
+    public synchronized String system_add_column_family(CfDef cf_def) throws InvalidRequestException,
TException
     {
         logger.debug("add_column_family");
         state().hasColumnFamilyListAccess(Permission.WRITE);
         ThriftValidation.validateCfDef(cf_def);
+        validateSchemaAgreement();
+
         try
         {
             applyMigrationOnStage(new AddColumnFamily(convertToCFMetaData(cf_def)));
@@ -685,10 +689,11 @@ public class CassandraServer implements 
         }
     }
 
-    public String system_drop_column_family(String column_family) throws InvalidRequestException,
TException
+    public synchronized String system_drop_column_family(String column_family) throws InvalidRequestException,
TException
     {
         logger.debug("drop_column_family");
         state().hasColumnFamilyListAccess(Permission.WRITE);
+        validateSchemaAgreement();
         
         try
         {
@@ -709,10 +714,11 @@ public class CassandraServer implements 
         }
     }
 
-    public String system_add_keyspace(KsDef ks_def) throws InvalidRequestException, TException
+    public synchronized String system_add_keyspace(KsDef ks_def) throws InvalidRequestException,
TException
     {
         logger.debug("add_keyspace");
         state().hasKeyspaceListAccess(Permission.WRITE);
+        validateSchemaAgreement();
         
         // generate a meaningful error if the user setup keyspace and/or column definition
incorrectly
         for (CfDef cf : ks_def.cf_defs) 
@@ -754,10 +760,11 @@ public class CassandraServer implements 
         }
     }
     
-    public String system_drop_keyspace(String keyspace) throws InvalidRequestException, TException
+    public synchronized String system_drop_keyspace(String keyspace) throws InvalidRequestException,
TException
     {
         logger.debug("drop_keyspace");
         state().hasKeyspaceListAccess(Permission.WRITE);
+        validateSchemaAgreement();
         
         try
         {
@@ -779,15 +786,15 @@ public class CassandraServer implements 
     }
 
     /** update an existing keyspace, but do not allow column family modifications. */
-    public String system_update_keyspace(KsDef ks_def) throws InvalidRequestException, TException
+    public synchronized String system_update_keyspace(KsDef ks_def) throws InvalidRequestException,
TException
     {
         logger.debug("update_keyspace");
         state().hasKeyspaceListAccess(Permission.WRITE);
-
         ThriftValidation.validateTable(ks_def.name);
         if (ks_def.getCf_defs() != null && ks_def.getCf_defs().size() > 0)
             throw new InvalidRequestException("Keyspace update must not contain any column
family definitions.");
-        
+        validateSchemaAgreement();
+
         try
         {
             KSMetaData ksm = new KSMetaData(
@@ -812,18 +819,17 @@ public class CassandraServer implements 
         }
     }
 
-    public String system_update_column_family(CfDef cf_def) throws InvalidRequestException,
TException
+    public synchronized 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)
             throw new InvalidRequestException("Keyspace and CF name must be set.");
-        
         CFMetaData oldCfm = DatabaseDescriptor.getCFMetaData(CFMetaData.getId(cf_def.keyspace,
cf_def.name));
         if (oldCfm == null) 
             throw new InvalidRequestException("Could not find column family definition to
modify.");
-        
+        validateSchemaAgreement();
+
         try
         {
             // ideally, apply() would happen on the stage with the
@@ -846,6 +852,15 @@ public class CassandraServer implements 
         }
     }
 
+    private void validateSchemaAgreement() throws InvalidRequestException
+    {
+        // unreachable hosts don't count towards disagreement
+        Map<String, List<String>> versions = Maps.filterKeys(StorageProxy.describeSchemaVersions(),
+                                                             Predicates.not(Predicates.equalTo(StorageProxy.UNREACHABLE)));
+        if (versions.size() > 1)
+            throw new InvalidRequestException("Cluster schema does not yet agree");
+    }
+
     // @see CFMetaData.applyImplicitDefaults().
     private CFMetaData convertToCFMetaData(CfDef cf_def) throws InvalidRequestException,
ConfigurationException
     {



Mime
View raw message