cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gdusba...@apache.org
Subject svn commit: r946717 - in /cassandra/trunk: interface/ interface/thrift/gen-java/org/apache/cassandra/thrift/ src/java/org/apache/cassandra/avro/ src/java/org/apache/cassandra/config/ src/java/org/apache/cassandra/db/ src/java/org/apache/cassandra/servi...
Date Thu, 20 May 2010 17:13:18 GMT
Author: gdusbabek
Date: Thu May 20 17:13:17 2010
New Revision: 946717

URL: http://svn.apache.org/viewvc?rev=946717&view=rev
Log:
service call to check for schema agreement. patch by gdusbabek, reviewed by jbellis. CASSANDRA-1075

Added:
    cassandra/trunk/src/java/org/apache/cassandra/db/SchemaCheckVerbHandler.java
Modified:
    cassandra/trunk/interface/cassandra.thrift
    cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
    cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java
    cassandra/trunk/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
    cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
    cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
    cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java

Modified: cassandra/trunk/interface/cassandra.thrift
URL: http://svn.apache.org/viewvc/cassandra/trunk/interface/cassandra.thrift?rev=946717&r1=946716&r2=946717&view=diff
==============================================================================
--- cassandra/trunk/interface/cassandra.thrift (original)
+++ cassandra/trunk/interface/cassandra.thrift Thu May 20 17:13:17 2010
@@ -435,6 +435,14 @@ service Cassandra {
     
   // Meta-APIs -- APIs to get information about the node or cluster,
   // rather than user data.  The nodeprobe program provides usage examples.
+  
+  /** 
+   * ask the cluster if they all are using the same migration id. returns a map of version->hosts-on-that-version.
+   * hosts that did not respond will be under the key DatabaseDescriptor.INITIAL_VERSION.
agreement can be determined
+   * by checking if the size of the map is 1. 
+   */
+  map<string, list<string>> check_schema_agreement()
+       throws (1: InvalidRequestException ire),
 
   /** list the defined keyspaces in this cluster */
   set<string> describe_keyspaces(),

Modified: cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java?rev=946717&r1=946716&r2=946717&view=diff
==============================================================================
--- cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java (original)
+++ cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java Thu
May 20 17:13:17 2010
@@ -162,6 +162,13 @@ public class Cassandra {
     public void truncate(String keyspace, String cfname) throws InvalidRequestException,
UnavailableException, TException;
 
     /**
+     * ask the cluster if they all are using the same migration id. returns a map of version->hosts-on-that-version.
+     * hosts that did not respond will be under the key DatabaseDescriptor.INITIAL_VERSION.
agreement can be determined
+     * by checking if the size of the map is 1.
+     */
+    public Map<String,List<String>> check_schema_agreement() throws InvalidRequestException,
TException;
+
+    /**
      * list the defined keyspaces in this cluster
      */
     public Set<String> describe_keyspaces() throws TException;
@@ -757,6 +764,41 @@ public class Cassandra {
       return;
     }
 
+    public Map<String,List<String>> check_schema_agreement() throws InvalidRequestException,
TException
+    {
+      send_check_schema_agreement();
+      return recv_check_schema_agreement();
+    }
+
+    public void send_check_schema_agreement() throws TException
+    {
+      oprot_.writeMessageBegin(new TMessage("check_schema_agreement", TMessageType.CALL,
seqid_));
+      check_schema_agreement_args args = new check_schema_agreement_args();
+      args.write(oprot_);
+      oprot_.writeMessageEnd();
+      oprot_.getTransport().flush();
+    }
+
+    public Map<String,List<String>> recv_check_schema_agreement() throws InvalidRequestException,
TException
+    {
+      TMessage msg = iprot_.readMessageBegin();
+      if (msg.type == TMessageType.EXCEPTION) {
+        TApplicationException x = TApplicationException.read(iprot_);
+        iprot_.readMessageEnd();
+        throw x;
+      }
+      check_schema_agreement_result result = new check_schema_agreement_result();
+      result.read(iprot_);
+      iprot_.readMessageEnd();
+      if (result.isSetSuccess()) {
+        return result.success;
+      }
+      if (result.ire != null) {
+        throw result.ire;
+      }
+      throw new TApplicationException(TApplicationException.MISSING_RESULT, "check_schema_agreement
failed: unknown result");
+    }
+
     public Set<String> describe_keyspaces() throws TException
     {
       send_describe_keyspaces();
@@ -1177,6 +1219,7 @@ public class Cassandra {
       processMap_.put("remove", new remove());
       processMap_.put("batch_mutate", new batch_mutate());
       processMap_.put("truncate", new truncate());
+      processMap_.put("check_schema_agreement", new check_schema_agreement());
       processMap_.put("describe_keyspaces", new describe_keyspaces());
       processMap_.put("describe_cluster_name", new describe_cluster_name());
       processMap_.put("describe_version", new describe_version());
@@ -1715,6 +1758,44 @@ public class Cassandra {
 
     }
 
+    private class check_schema_agreement implements ProcessFunction {
+      public void process(int seqid, TProtocol iprot, TProtocol oprot) throws TException
+      {
+        check_schema_agreement_args args = new check_schema_agreement_args();
+        try {
+          args.read(iprot);
+        } catch (TProtocolException e) {
+          iprot.readMessageEnd();
+          TApplicationException x = new TApplicationException(TApplicationException.PROTOCOL_ERROR,
e.getMessage());
+          oprot.writeMessageBegin(new TMessage("check_schema_agreement", TMessageType.EXCEPTION,
seqid));
+          x.write(oprot);
+          oprot.writeMessageEnd();
+          oprot.getTransport().flush();
+          return;
+        }
+        iprot.readMessageEnd();
+        check_schema_agreement_result result = new check_schema_agreement_result();
+        try {
+          result.success = iface_.check_schema_agreement();
+        } catch (InvalidRequestException ire) {
+          result.ire = ire;
+        } catch (Throwable th) {
+          LOGGER.error("Internal error processing check_schema_agreement", th);
+          TApplicationException x = new TApplicationException(TApplicationException.INTERNAL_ERROR,
"Internal error processing check_schema_agreement");
+          oprot.writeMessageBegin(new TMessage("check_schema_agreement", TMessageType.EXCEPTION,
seqid));
+          x.write(oprot);
+          oprot.writeMessageEnd();
+          oprot.getTransport().flush();
+          return;
+        }
+        oprot.writeMessageBegin(new TMessage("check_schema_agreement", TMessageType.REPLY,
seqid));
+        result.write(oprot);
+        oprot.writeMessageEnd();
+        oprot.getTransport().flush();
+      }
+
+    }
+
     private class describe_keyspaces implements ProcessFunction {
       public void process(int seqid, TProtocol iprot, TProtocol oprot) throws TException
       {
@@ -13999,6 +14080,598 @@ public class Cassandra {
 
   }
 
+  public static class check_schema_agreement_args implements TBase<check_schema_agreement_args._Fields>,
java.io.Serializable, Cloneable, Comparable<check_schema_agreement_args>   {
+    private static final TStruct STRUCT_DESC = new TStruct("check_schema_agreement_args");
+
+
+
+    /** The set of fields this struct contains, along with convenience methods for finding
and manipulating them. */
+    public enum _Fields implements TFieldIdEnum {
+;
+
+      private static final Map<Integer, _Fields> byId = new HashMap<Integer, _Fields>();
+      private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+      static {
+        for (_Fields field : EnumSet.allOf(_Fields.class)) {
+          byId.put((int)field._thriftId, field);
+          byName.put(field.getFieldName(), field);
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, or null if its not found.
+       */
+      public static _Fields findByThriftId(int fieldId) {
+        return byId.get(fieldId);
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, throwing an exception
+       * if it is not found.
+       */
+      public static _Fields findByThriftIdOrThrow(int fieldId) {
+        _Fields fields = findByThriftId(fieldId);
+        if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't
exist!");
+        return fields;
+      }
+
+      /**
+       * Find the _Fields constant that matches name, or null if its not found.
+       */
+      public static _Fields findByName(String name) {
+        return byName.get(name);
+      }
+
+      private final short _thriftId;
+      private final String _fieldName;
+
+      _Fields(short thriftId, String fieldName) {
+        _thriftId = thriftId;
+        _fieldName = fieldName;
+      }
+
+      public short getThriftFieldId() {
+        return _thriftId;
+      }
+
+      public String getFieldName() {
+        return _fieldName;
+      }
+    }
+    public static final Map<_Fields, FieldMetaData> metaDataMap = Collections.unmodifiableMap(new
EnumMap<_Fields, FieldMetaData>(_Fields.class) {{
+    }});
+
+    static {
+      FieldMetaData.addStructMetaDataMap(check_schema_agreement_args.class, metaDataMap);
+    }
+
+    public check_schema_agreement_args() {
+    }
+
+    /**
+     * Performs a deep copy on <i>other</i>.
+     */
+    public check_schema_agreement_args(check_schema_agreement_args other) {
+    }
+
+    public check_schema_agreement_args deepCopy() {
+      return new check_schema_agreement_args(this);
+    }
+
+    @Deprecated
+    public check_schema_agreement_args clone() {
+      return new check_schema_agreement_args(this);
+    }
+
+    public void setFieldValue(_Fields field, Object value) {
+      switch (field) {
+      }
+    }
+
+    public void setFieldValue(int fieldID, Object value) {
+      setFieldValue(_Fields.findByThriftIdOrThrow(fieldID), value);
+    }
+
+    public Object getFieldValue(_Fields field) {
+      switch (field) {
+      }
+      throw new IllegalStateException();
+    }
+
+    public Object getFieldValue(int fieldId) {
+      return getFieldValue(_Fields.findByThriftIdOrThrow(fieldId));
+    }
+
+    /** Returns true if field corresponding to fieldID is set (has been asigned a value)
and false otherwise */
+    public boolean isSet(_Fields field) {
+      switch (field) {
+      }
+      throw new IllegalStateException();
+    }
+
+    public boolean isSet(int fieldID) {
+      return isSet(_Fields.findByThriftIdOrThrow(fieldID));
+    }
+
+    @Override
+    public boolean equals(Object that) {
+      if (that == null)
+        return false;
+      if (that instanceof check_schema_agreement_args)
+        return this.equals((check_schema_agreement_args)that);
+      return false;
+    }
+
+    public boolean equals(check_schema_agreement_args that) {
+      if (that == null)
+        return false;
+
+      return true;
+    }
+
+    @Override
+    public int hashCode() {
+      return 0;
+    }
+
+    public int compareTo(check_schema_agreement_args other) {
+      if (!getClass().equals(other.getClass())) {
+        return getClass().getName().compareTo(other.getClass().getName());
+      }
+
+      int lastComparison = 0;
+      check_schema_agreement_args typedOther = (check_schema_agreement_args)other;
+
+      return 0;
+    }
+
+    public void read(TProtocol iprot) throws TException {
+      TField field;
+      iprot.readStructBegin();
+      while (true)
+      {
+        field = iprot.readFieldBegin();
+        if (field.type == TType.STOP) { 
+          break;
+        }
+        switch (field.id) {
+          default:
+            TProtocolUtil.skip(iprot, field.type);
+        }
+        iprot.readFieldEnd();
+      }
+      iprot.readStructEnd();
+
+      // check for required fields of primitive type, which can't be checked in the validate
method
+      validate();
+    }
+
+    public void write(TProtocol oprot) throws TException {
+      validate();
+
+      oprot.writeStructBegin(STRUCT_DESC);
+      oprot.writeFieldStop();
+      oprot.writeStructEnd();
+    }
+
+    @Override
+    public String toString() {
+      StringBuilder sb = new StringBuilder("check_schema_agreement_args(");
+      boolean first = true;
+
+      sb.append(")");
+      return sb.toString();
+    }
+
+    public void validate() throws TException {
+      // check for required fields
+    }
+
+  }
+
+  public static class check_schema_agreement_result implements TBase<check_schema_agreement_result._Fields>,
java.io.Serializable, Cloneable   {
+    private static final TStruct STRUCT_DESC = new TStruct("check_schema_agreement_result");
+
+    private static final TField SUCCESS_FIELD_DESC = new TField("success", TType.MAP, (short)0);
+    private static final TField IRE_FIELD_DESC = new TField("ire", TType.STRUCT, (short)1);
+
+    public Map<String,List<String>> success;
+    public InvalidRequestException ire;
+
+    /** The set of fields this struct contains, along with convenience methods for finding
and manipulating them. */
+    public enum _Fields implements TFieldIdEnum {
+      SUCCESS((short)0, "success"),
+      IRE((short)1, "ire");
+
+      private static final Map<Integer, _Fields> byId = new HashMap<Integer, _Fields>();
+      private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+      static {
+        for (_Fields field : EnumSet.allOf(_Fields.class)) {
+          byId.put((int)field._thriftId, field);
+          byName.put(field.getFieldName(), field);
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, or null if its not found.
+       */
+      public static _Fields findByThriftId(int fieldId) {
+        return byId.get(fieldId);
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, throwing an exception
+       * if it is not found.
+       */
+      public static _Fields findByThriftIdOrThrow(int fieldId) {
+        _Fields fields = findByThriftId(fieldId);
+        if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't
exist!");
+        return fields;
+      }
+
+      /**
+       * Find the _Fields constant that matches name, or null if its not found.
+       */
+      public static _Fields findByName(String name) {
+        return byName.get(name);
+      }
+
+      private final short _thriftId;
+      private final String _fieldName;
+
+      _Fields(short thriftId, String fieldName) {
+        _thriftId = thriftId;
+        _fieldName = fieldName;
+      }
+
+      public short getThriftFieldId() {
+        return _thriftId;
+      }
+
+      public String getFieldName() {
+        return _fieldName;
+      }
+    }
+
+    // isset id assignments
+
+    public static final Map<_Fields, FieldMetaData> metaDataMap = Collections.unmodifiableMap(new
EnumMap<_Fields, FieldMetaData>(_Fields.class) {{
+      put(_Fields.SUCCESS, new FieldMetaData("success", TFieldRequirementType.DEFAULT, 
+          new MapMetaData(TType.MAP, 
+              new FieldValueMetaData(TType.STRING), 
+              new ListMetaData(TType.LIST, 
+                  new FieldValueMetaData(TType.STRING)))));
+      put(_Fields.IRE, new FieldMetaData("ire", TFieldRequirementType.DEFAULT, 
+          new FieldValueMetaData(TType.STRUCT)));
+    }});
+
+    static {
+      FieldMetaData.addStructMetaDataMap(check_schema_agreement_result.class, metaDataMap);
+    }
+
+    public check_schema_agreement_result() {
+    }
+
+    public check_schema_agreement_result(
+      Map<String,List<String>> success,
+      InvalidRequestException ire)
+    {
+      this();
+      this.success = success;
+      this.ire = ire;
+    }
+
+    /**
+     * Performs a deep copy on <i>other</i>.
+     */
+    public check_schema_agreement_result(check_schema_agreement_result other) {
+      if (other.isSetSuccess()) {
+        Map<String,List<String>> __this__success = new HashMap<String,List<String>>();
+        for (Map.Entry<String, List<String>> other_element : other.success.entrySet())
{
+
+          String other_element_key = other_element.getKey();
+          List<String> other_element_value = other_element.getValue();
+
+          String __this__success_copy_key = other_element_key;
+
+          List<String> __this__success_copy_value = new ArrayList<String>();
+          for (String other_element_value_element : other_element_value) {
+            __this__success_copy_value.add(other_element_value_element);
+          }
+
+          __this__success.put(__this__success_copy_key, __this__success_copy_value);
+        }
+        this.success = __this__success;
+      }
+      if (other.isSetIre()) {
+        this.ire = new InvalidRequestException(other.ire);
+      }
+    }
+
+    public check_schema_agreement_result deepCopy() {
+      return new check_schema_agreement_result(this);
+    }
+
+    @Deprecated
+    public check_schema_agreement_result clone() {
+      return new check_schema_agreement_result(this);
+    }
+
+    public int getSuccessSize() {
+      return (this.success == null) ? 0 : this.success.size();
+    }
+
+    public void putToSuccess(String key, List<String> val) {
+      if (this.success == null) {
+        this.success = new HashMap<String,List<String>>();
+      }
+      this.success.put(key, val);
+    }
+
+    public Map<String,List<String>> getSuccess() {
+      return this.success;
+    }
+
+    public check_schema_agreement_result setSuccess(Map<String,List<String>>
success) {
+      this.success = success;
+      return this;
+    }
+
+    public void unsetSuccess() {
+      this.success = null;
+    }
+
+    /** Returns true if field success is set (has been asigned a value) and false otherwise
*/
+    public boolean isSetSuccess() {
+      return this.success != null;
+    }
+
+    public void setSuccessIsSet(boolean value) {
+      if (!value) {
+        this.success = null;
+      }
+    }
+
+    public InvalidRequestException getIre() {
+      return this.ire;
+    }
+
+    public check_schema_agreement_result setIre(InvalidRequestException ire) {
+      this.ire = ire;
+      return this;
+    }
+
+    public void unsetIre() {
+      this.ire = null;
+    }
+
+    /** Returns true if field ire is set (has been asigned a value) and false otherwise */
+    public boolean isSetIre() {
+      return this.ire != null;
+    }
+
+    public void setIreIsSet(boolean value) {
+      if (!value) {
+        this.ire = null;
+      }
+    }
+
+    public void setFieldValue(_Fields field, Object value) {
+      switch (field) {
+      case SUCCESS:
+        if (value == null) {
+          unsetSuccess();
+        } else {
+          setSuccess((Map<String,List<String>>)value);
+        }
+        break;
+
+      case IRE:
+        if (value == null) {
+          unsetIre();
+        } else {
+          setIre((InvalidRequestException)value);
+        }
+        break;
+
+      }
+    }
+
+    public void setFieldValue(int fieldID, Object value) {
+      setFieldValue(_Fields.findByThriftIdOrThrow(fieldID), value);
+    }
+
+    public Object getFieldValue(_Fields field) {
+      switch (field) {
+      case SUCCESS:
+        return getSuccess();
+
+      case IRE:
+        return getIre();
+
+      }
+      throw new IllegalStateException();
+    }
+
+    public Object getFieldValue(int fieldId) {
+      return getFieldValue(_Fields.findByThriftIdOrThrow(fieldId));
+    }
+
+    /** Returns true if field corresponding to fieldID is set (has been asigned a value)
and false otherwise */
+    public boolean isSet(_Fields field) {
+      switch (field) {
+      case SUCCESS:
+        return isSetSuccess();
+      case IRE:
+        return isSetIre();
+      }
+      throw new IllegalStateException();
+    }
+
+    public boolean isSet(int fieldID) {
+      return isSet(_Fields.findByThriftIdOrThrow(fieldID));
+    }
+
+    @Override
+    public boolean equals(Object that) {
+      if (that == null)
+        return false;
+      if (that instanceof check_schema_agreement_result)
+        return this.equals((check_schema_agreement_result)that);
+      return false;
+    }
+
+    public boolean equals(check_schema_agreement_result that) {
+      if (that == null)
+        return false;
+
+      boolean this_present_success = true && this.isSetSuccess();
+      boolean that_present_success = true && that.isSetSuccess();
+      if (this_present_success || that_present_success) {
+        if (!(this_present_success && that_present_success))
+          return false;
+        if (!this.success.equals(that.success))
+          return false;
+      }
+
+      boolean this_present_ire = true && this.isSetIre();
+      boolean that_present_ire = true && that.isSetIre();
+      if (this_present_ire || that_present_ire) {
+        if (!(this_present_ire && that_present_ire))
+          return false;
+        if (!this.ire.equals(that.ire))
+          return false;
+      }
+
+      return true;
+    }
+
+    @Override
+    public int hashCode() {
+      return 0;
+    }
+
+    public void read(TProtocol iprot) throws TException {
+      TField field;
+      iprot.readStructBegin();
+      while (true)
+      {
+        field = iprot.readFieldBegin();
+        if (field.type == TType.STOP) { 
+          break;
+        }
+        switch (field.id) {
+          case 0: // SUCCESS
+            if (field.type == TType.MAP) {
+              {
+                TMap _map73 = iprot.readMapBegin();
+                this.success = new HashMap<String,List<String>>(2*_map73.size);
+                for (int _i74 = 0; _i74 < _map73.size; ++_i74)
+                {
+                  String _key75;
+                  List<String> _val76;
+                  _key75 = iprot.readString();
+                  {
+                    TList _list77 = iprot.readListBegin();
+                    _val76 = new ArrayList<String>(_list77.size);
+                    for (int _i78 = 0; _i78 < _list77.size; ++_i78)
+                    {
+                      String _elem79;
+                      _elem79 = iprot.readString();
+                      _val76.add(_elem79);
+                    }
+                    iprot.readListEnd();
+                  }
+                  this.success.put(_key75, _val76);
+                }
+                iprot.readMapEnd();
+              }
+            } else { 
+              TProtocolUtil.skip(iprot, field.type);
+            }
+            break;
+          case 1: // IRE
+            if (field.type == TType.STRUCT) {
+              this.ire = new InvalidRequestException();
+              this.ire.read(iprot);
+            } else { 
+              TProtocolUtil.skip(iprot, field.type);
+            }
+            break;
+          default:
+            TProtocolUtil.skip(iprot, field.type);
+        }
+        iprot.readFieldEnd();
+      }
+      iprot.readStructEnd();
+
+      // check for required fields of primitive type, which can't be checked in the validate
method
+      validate();
+    }
+
+    public void write(TProtocol oprot) throws TException {
+      oprot.writeStructBegin(STRUCT_DESC);
+
+      if (this.isSetSuccess()) {
+        oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
+        {
+          oprot.writeMapBegin(new TMap(TType.STRING, TType.LIST, this.success.size()));
+          for (Map.Entry<String, List<String>> _iter80 : this.success.entrySet())
+          {
+            oprot.writeString(_iter80.getKey());
+            {
+              oprot.writeListBegin(new TList(TType.STRING, _iter80.getValue().size()));
+              for (String _iter81 : _iter80.getValue())
+              {
+                oprot.writeString(_iter81);
+              }
+              oprot.writeListEnd();
+            }
+          }
+          oprot.writeMapEnd();
+        }
+        oprot.writeFieldEnd();
+      } else if (this.isSetIre()) {
+        oprot.writeFieldBegin(IRE_FIELD_DESC);
+        this.ire.write(oprot);
+        oprot.writeFieldEnd();
+      }
+      oprot.writeFieldStop();
+      oprot.writeStructEnd();
+    }
+
+    @Override
+    public String toString() {
+      StringBuilder sb = new StringBuilder("check_schema_agreement_result(");
+      boolean first = true;
+
+      sb.append("success:");
+      if (this.success == null) {
+        sb.append("null");
+      } else {
+        sb.append(this.success);
+      }
+      first = false;
+      if (!first) sb.append(", ");
+      sb.append("ire:");
+      if (this.ire == null) {
+        sb.append("null");
+      } else {
+        sb.append(this.ire);
+      }
+      first = false;
+      sb.append(")");
+      return sb.toString();
+    }
+
+    public void validate() throws TException {
+      // check for required fields
+    }
+
+  }
+
   public static class describe_keyspaces_args implements TBase<describe_keyspaces_args._Fields>,
java.io.Serializable, Cloneable, Comparable<describe_keyspaces_args>   {
     private static final TStruct STRUCT_DESC = new TStruct("describe_keyspaces_args");
 
@@ -14420,13 +15093,13 @@ public class Cassandra {
           case 0: // SUCCESS
             if (field.type == TType.SET) {
               {
-                TSet _set73 = iprot.readSetBegin();
-                this.success = new HashSet<String>(2*_set73.size);
-                for (int _i74 = 0; _i74 < _set73.size; ++_i74)
+                TSet _set82 = iprot.readSetBegin();
+                this.success = new HashSet<String>(2*_set82.size);
+                for (int _i83 = 0; _i83 < _set82.size; ++_i83)
                 {
-                  String _elem75;
-                  _elem75 = iprot.readString();
-                  this.success.add(_elem75);
+                  String _elem84;
+                  _elem84 = iprot.readString();
+                  this.success.add(_elem84);
                 }
                 iprot.readSetEnd();
               }
@@ -14452,9 +15125,9 @@ public class Cassandra {
         oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
         {
           oprot.writeSetBegin(new TSet(TType.STRING, this.success.size()));
-          for (String _iter76 : this.success)
+          for (String _iter85 : this.success)
           {
-            oprot.writeString(_iter76);
+            oprot.writeString(_iter85);
           }
           oprot.writeSetEnd();
         }
@@ -15961,14 +16634,14 @@ public class Cassandra {
           case 0: // SUCCESS
             if (field.type == TType.LIST) {
               {
-                TList _list77 = iprot.readListBegin();
-                this.success = new ArrayList<TokenRange>(_list77.size);
-                for (int _i78 = 0; _i78 < _list77.size; ++_i78)
+                TList _list86 = iprot.readListBegin();
+                this.success = new ArrayList<TokenRange>(_list86.size);
+                for (int _i87 = 0; _i87 < _list86.size; ++_i87)
                 {
-                  TokenRange _elem79;
-                  _elem79 = new TokenRange();
-                  _elem79.read(iprot);
-                  this.success.add(_elem79);
+                  TokenRange _elem88;
+                  _elem88 = new TokenRange();
+                  _elem88.read(iprot);
+                  this.success.add(_elem88);
                 }
                 iprot.readListEnd();
               }
@@ -15994,9 +16667,9 @@ public class Cassandra {
         oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
         {
           oprot.writeListBegin(new TList(TType.STRUCT, this.success.size()));
-          for (TokenRange _iter80 : this.success)
+          for (TokenRange _iter89 : this.success)
           {
-            _iter80.write(oprot);
+            _iter89.write(oprot);
           }
           oprot.writeListEnd();
         }
@@ -16617,27 +17290,27 @@ public class Cassandra {
           case 0: // SUCCESS
             if (field.type == TType.MAP) {
               {
-                TMap _map81 = iprot.readMapBegin();
-                this.success = new HashMap<String,Map<String,String>>(2*_map81.size);
-                for (int _i82 = 0; _i82 < _map81.size; ++_i82)
+                TMap _map90 = iprot.readMapBegin();
+                this.success = new HashMap<String,Map<String,String>>(2*_map90.size);
+                for (int _i91 = 0; _i91 < _map90.size; ++_i91)
                 {
-                  String _key83;
-                  Map<String,String> _val84;
-                  _key83 = iprot.readString();
+                  String _key92;
+                  Map<String,String> _val93;
+                  _key92 = iprot.readString();
                   {
-                    TMap _map85 = iprot.readMapBegin();
-                    _val84 = new HashMap<String,String>(2*_map85.size);
-                    for (int _i86 = 0; _i86 < _map85.size; ++_i86)
+                    TMap _map94 = iprot.readMapBegin();
+                    _val93 = new HashMap<String,String>(2*_map94.size);
+                    for (int _i95 = 0; _i95 < _map94.size; ++_i95)
                     {
-                      String _key87;
-                      String _val88;
-                      _key87 = iprot.readString();
-                      _val88 = iprot.readString();
-                      _val84.put(_key87, _val88);
+                      String _key96;
+                      String _val97;
+                      _key96 = iprot.readString();
+                      _val97 = iprot.readString();
+                      _val93.put(_key96, _val97);
                     }
                     iprot.readMapEnd();
                   }
-                  this.success.put(_key83, _val84);
+                  this.success.put(_key92, _val93);
                 }
                 iprot.readMapEnd();
               }
@@ -16671,15 +17344,15 @@ public class Cassandra {
         oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
         {
           oprot.writeMapBegin(new TMap(TType.STRING, TType.MAP, this.success.size()));
-          for (Map.Entry<String, Map<String,String>> _iter89 : this.success.entrySet())
+          for (Map.Entry<String, Map<String,String>> _iter98 : this.success.entrySet())
           {
-            oprot.writeString(_iter89.getKey());
+            oprot.writeString(_iter98.getKey());
             {
-              oprot.writeMapBegin(new TMap(TType.STRING, TType.STRING, _iter89.getValue().size()));
-              for (Map.Entry<String, String> _iter90 : _iter89.getValue().entrySet())
+              oprot.writeMapBegin(new TMap(TType.STRING, TType.STRING, _iter98.getValue().size()));
+              for (Map.Entry<String, String> _iter99 : _iter98.getValue().entrySet())
               {
-                oprot.writeString(_iter90.getKey());
-                oprot.writeString(_iter90.getValue());
+                oprot.writeString(_iter99.getKey());
+                oprot.writeString(_iter99.getValue());
               }
               oprot.writeMapEnd();
             }
@@ -17435,13 +18108,13 @@ public class Cassandra {
           case 0: // SUCCESS
             if (field.type == TType.LIST) {
               {
-                TList _list91 = iprot.readListBegin();
-                this.success = new ArrayList<String>(_list91.size);
-                for (int _i92 = 0; _i92 < _list91.size; ++_i92)
+                TList _list100 = iprot.readListBegin();
+                this.success = new ArrayList<String>(_list100.size);
+                for (int _i101 = 0; _i101 < _list100.size; ++_i101)
                 {
-                  String _elem93;
-                  _elem93 = iprot.readString();
-                  this.success.add(_elem93);
+                  String _elem102;
+                  _elem102 = iprot.readString();
+                  this.success.add(_elem102);
                 }
                 iprot.readListEnd();
               }
@@ -17467,9 +18140,9 @@ public class Cassandra {
         oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
         {
           oprot.writeListBegin(new TList(TType.STRING, this.success.size()));
-          for (String _iter94 : this.success)
+          for (String _iter103 : this.success)
           {
-            oprot.writeString(_iter94);
+            oprot.writeString(_iter103);
           }
           oprot.writeListEnd();
         }

Modified: cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java?rev=946717&r1=946716&r2=946717&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/avro/CassandraServer.java Thu May 20 17:13:17
2010
@@ -51,6 +51,8 @@ import org.apache.cassandra.service.Stor
 import static org.apache.cassandra.utils.FBUtilities.UTF8;
 
 import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.thrift.*;
+import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import static org.apache.cassandra.avro.AvroRecordFactory.*;
@@ -573,4 +575,10 @@ public class CassandraServer implements 
     {
         return API_VERSION;
     }
+    
+    public Map<String, List<String>> check_schema_agreement()
+    {
+        logger.debug("checking schema agreement");      
+        return StorageProxy.checkSchemaAgreement();
+    }
 }

Modified: cassandra/trunk/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/config/DatabaseDescriptor.java?rev=946717&r1=946716&r2=946717&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/config/DatabaseDescriptor.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/config/DatabaseDescriptor.java Thu May 20
17:13:17 2010
@@ -84,7 +84,7 @@ public class DatabaseDescriptor
 
     private final static String STORAGE_CONF_FILE = "cassandra.yaml";
 
-    private static final UUID INITIAL_VERSION = new UUID(4096, 0); // has type nibble set
to 1, everything else to zero.
+    public static final UUID INITIAL_VERSION = new UUID(4096, 0); // has type nibble set
to 1, everything else to zero.
     private static UUID defsVersion = INITIAL_VERSION;
 
     /**

Added: cassandra/trunk/src/java/org/apache/cassandra/db/SchemaCheckVerbHandler.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/SchemaCheckVerbHandler.java?rev=946717&view=auto
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/SchemaCheckVerbHandler.java (added)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/SchemaCheckVerbHandler.java Thu May 20
17:13:17 2010
@@ -0,0 +1,41 @@
+package org.apache.cassandra.db;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.IVerbHandler;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.FBUtilities;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+
+
+public class SchemaCheckVerbHandler implements IVerbHandler
+{
+    private final Logger logger = LoggerFactory.getLogger(SchemaCheckVerbHandler.class);
+    
+    @Override
+    public void doVerb(Message message)
+    {
+        logger.debug("Received schema check request.");
+        Message response = message.getReply(FBUtilities.getLocalAddress(), DatabaseDescriptor.getDefsVersion().toString().getBytes());
+        MessagingService.instance.sendOneWay(response, message.getFrom());
+    }
+}

Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java?rev=946717&r1=946716&r2=946717&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java Thu May 20 17:13:17
2010
@@ -26,12 +26,17 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -55,6 +60,7 @@ import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.gms.Gossiper;
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
 import org.apache.cassandra.locator.TokenMetadata;
+import org.apache.cassandra.net.IAsyncCallback;
 import org.apache.cassandra.net.IAsyncResult;
 import org.apache.cassandra.net.Message;
 import org.apache.cassandra.net.MessagingService;
@@ -565,6 +571,76 @@ public class StorageProxy implements Sto
     }
 
     /**
+     * initiate a request/response session with each live node to check whether or not everybody
is using the same 
+     * migration id. This is useful for determining if a schema change has propagated through
the cluster. Disagreement
+     * is assumed if any node fails to respond.
+     */
+    public static Map<String, List<String>> checkSchemaAgreement()
+    {
+        final Map<String, List<String>> results = new HashMap<String, List<String>>();
+        
+        final String myVersion = DatabaseDescriptor.getDefsVersion().toString();
+        final Map<InetAddress, UUID> versions = new ConcurrentHashMap<InetAddress,
UUID>();
+        final Set<InetAddress> liveHosts = Gossiper.instance.getLiveMembers();
+        final Message msg = new Message(FBUtilities.getLocalAddress(), StageManager.MIGRATION_STAGE,
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() 
+        {
+            @Override
+            public void response(Message msg)
+            {
+                // record the response from the remote node.
+                logger.debug("Received schema check response from " + msg.getFrom().getHostAddress());
+                UUID theirVersion = UUID.fromString(new String(msg.getMessageBody()));
+                versions.put(msg.getFrom(), theirVersion);
+                latch.countDown();
+            }
+        });
+        
+        try
+        {
+            // wait for as long as possible. timeout-1s if possible.
+            latch.await(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
+        } 
+        catch (InterruptedException ex) 
+        {
+            throw new AssertionError("This latch shouldn't have been interrupted.");
+        }
+        
+        logger.debug("My version is " + myVersion);
+        
+        // first, indicate any hosts that did not respond.
+        final Set<InetAddress> ackedHosts = versions.keySet();
+        if (ackedHosts.size() < liveHosts.size())
+        {
+            Set<InetAddress> missingHosts = new HashSet<InetAddress>(liveHosts);
+            missingHosts.removeAll(ackedHosts);
+            assert missingHosts.size() > 0;
+            List<String> missingHostNames = new ArrayList<String>(missingHosts.size());
+            for (InetAddress host : missingHosts)
+                missingHostNames.add(host.getHostAddress());
+            results.put(DatabaseDescriptor.INITIAL_VERSION.toString(), missingHostNames);
+            logger.debug("Hosts not in agreement. Didn't get a response from everybody: "
+ StringUtils.join(missingHostNames, ","));
+        }
+        
+        // check for version disagreement. log the hosts that don't agree.
+        for (InetAddress host : ackedHosts)
+        {
+            String uuid = versions.get(host).toString();
+            if (!results.containsKey(uuid))
+                results.put(uuid, new ArrayList<String>());
+            results.get(uuid).add(host.getHostAddress());
+            if (!uuid.equals(myVersion))
+                logger.debug("%s disagrees (%s)", host.getHostAddress(), uuid);
+        }
+        if (results.size() == 1)
+            logger.debug("Schemas are in agreement.");
+        
+        return results;
+    }
+
+    /**
      * returns an iterator that will return ranges in ring order, starting with the one that
contains the start token
      */
     private static Iterable<Pair<AbstractBounds, List<InetAddress>>> getRangeIterator(final
List<Pair<AbstractBounds, List<InetAddress>>> ranges, Token start)

Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java?rev=946717&r1=946716&r2=946717&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java Thu May 20 17:13:17
2010
@@ -111,6 +111,7 @@ public class StorageService implements I
         DEFINITIONS_ANNOUNCE,
         DEFINITIONS_UPDATE_RESPONSE,
         TRUNCATE,
+        SCHEMA_CHECK,
         ;
         // remember to add new verbs at the end, since we serialize by ordinal
     }
@@ -236,6 +237,7 @@ public class StorageService implements I
         MessagingService.instance.registerVerbHandlers(Verb.DEFINITIONS_ANNOUNCE, new DefinitionsAnnounceVerbHandler());
         MessagingService.instance.registerVerbHandlers(Verb.DEFINITIONS_UPDATE_RESPONSE,
new DefinitionsUpdateResponseVerbHandler());
         MessagingService.instance.registerVerbHandlers(Verb.TRUNCATE, new TruncateVerbHandler());
+        MessagingService.instance.registerVerbHandlers(Verb.SCHEMA_CHECK, new SchemaCheckVerbHandler());
 
         replicationStrategies = new HashMap<String, AbstractReplicationStrategy>();
         for (String table : DatabaseDescriptor.getNonSystemTables())

Modified: cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java?rev=946717&r1=946716&r2=946717&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java Thu May 20 17:13:17
2010
@@ -856,5 +856,12 @@ public class CassandraServer implements 
         keySpace.set(keyspace); 
     }
 
+    @Override
+    public Map<String, List<String>> check_schema_agreement() throws TException,
InvalidRequestException
+    {
+        logger.debug("checking schema agreement");      
+        return StorageProxy.checkSchemaAgreement();
+    }
+
     // main method moved to CassandraDaemon
 }



Mime
View raw message