RUBY-254 - beta protocol support

* Added allow_beta_protocol option to Cassandra.cluster
* Added handling for v5 versions of read-failure and write-failure responses.
* Added ability to run integration tests against a C* cluster using a locally build C* tree.

Author: stamhankar999

CHANGELOG.md

@@ -4,6 +4,7 @@ Features:
 * Update Keyspace metadata to include collection of indexes defined in the keyspace.
 * Update Table metadata to include trigger-collection and view-collection metadata.
 * Added execution profiles to encapsulate a group of request execution options.
+* Added support for v5 beta protocol.
 
 Bug Fixes:
 * [RUBY-255](https://datastax-oss.atlassian.net/browse/RUBY-255) ControlConnection.peer_ip ignores peers that are missing critical information in system.peers.

integration/client_error_test.rb

@@ -69,32 +69,43 @@ def set_failing_nodes(failing_nodes, keyspace)
   # @test_category error_codes
   #
   def test_raise_error_on_write_failure
-    begin
-      skip("Client failure errors are only available in C* after 2.2") if CCM.cassandra_version < '2.2.0'
-
-      cluster = Retry.with_attempts(5, Cassandra::Errors::NoHostsAvailable) { Cassandra.cluster }
-      session = cluster.connect
+    skip("Client failure errors are only available in C* after 2.2") if CCM.cassandra_version < '2.2.0'
 
-      session.execute("CREATE KEYSPACE testwritefail WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}",
-                      consistency: :all) rescue nil
-      session.execute("CREATE TABLE testwritefail.test (k int PRIMARY KEY, v int)", consistency: :all)
+    # Create a cluster object that tries to use the beta protocol. If it succeeds, the write failure response
+    # will have a map of <node-ip, failure-code> instead of num-failures. When v5 is officially released, we
+    # can remove the allow_beta_protocol arg in this test.
+    cluster = Retry.with_attempts(5, Cassandra::Errors::NoHostsAvailable) {
+      Cassandra.cluster(allow_beta_protocol: true)
+    }
+    session = cluster.connect
 
-      # Disable one node
-      set_failing_nodes(["node1"], "testwritefail")
+    session.execute("CREATE KEYSPACE testwritefail WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}",
+                    consistency: :all) rescue nil
+    session.execute("CREATE TABLE IF NOT EXISTS testwritefail.test (k int PRIMARY KEY, v int)", consistency: :all)
 
-      # One node disabled should trigger a WriteFailure
-      assert_raises(Cassandra::Errors::WriteError) do
-        session.execute("INSERT INTO testwritefail.test (k, v) VALUES (1, 0)", consistency: :all)
-      end
+    # Disable one node
+    set_failing_nodes(["node1"], "testwritefail")
 
-      # Quorum should still work with two nodes
-      session.execute("INSERT INTO testwritefail.test (k, v) VALUES (1, 0)", consistency: :quorum)
+    # One node disabled should trigger a WriteFailure
+    ex = assert_raises(Cassandra::Errors::WriteError) do
+      session.execute("INSERT INTO testwritefail.test (k, v) VALUES (1, 0)", consistency: :all)
+    end
 
-      # Restart the node to clear jvm settings
-      set_failing_nodes([], "testwritefail")
-    ensure
-      cluster && cluster.close
+    # If we're speaking protocol version > 4, verify that we have a failure map in the exception.
+    connection_options = cluster.instance_variable_get(:@connection_options)
+    assert_operator(ex.failed, :>=, 1)
+    if connection_options.protocol_version > 4
+      assert_equal(ex.failed, ex.failures_by_node.size)
+      assert_equal(0, ex.failures_by_node.values.first)
     end
+
+    # Quorum should still work with two nodes
+    session.execute("INSERT INTO testwritefail.test (k, v) VALUES (1, 0)", consistency: :quorum)
+
+    # Restart the node to clear jvm settings
+    set_failing_nodes([], "testwritefail")
+  ensure
+    cluster && cluster.close
   end
 
   # Test for validating ReadError
@@ -117,33 +128,45 @@ def test_raise_error_on_write_failure
   def test_raise_error_on_read_failure
     skip("Client failure errors are only available in C* after 2.2") if CCM.cassandra_version < '2.2.0'
 
-    begin
-      cluster = Retry.with_attempts(5, Cassandra::Errors::NoHostsAvailable) { Cassandra.cluster }
-      session = cluster.connect
-
-      session.execute("CREATE KEYSPACE testreadfail WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}",
-                      consistency: :all) rescue nil
-      session.execute("CREATE TABLE testreadfail.test2 (k int, v0 int, v1 int, PRIMARY KEY (k, v0))", consistency: :all)
+    # Create a cluster object that tries to use the beta protocol. If it succeeds, the write failure response
+    # will have a map of <node-ip, failure-code> instead of num-failures. When v5 is officially released, we
+    # can remove the allow_beta_protocol arg in this test.
+    cluster = Retry.with_attempts(5, Cassandra::Errors::NoHostsAvailable) {
+      Cassandra.cluster(allow_beta_protocol: true)
+    }
+    session = cluster.connect
+
+    session.execute("CREATE KEYSPACE testreadfail WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}",
+                    consistency: :all) rescue nil
+    session.execute("CREATE TABLE IF NOT EXISTS testreadfail.test2 (k int, v0 int, v1 int, PRIMARY KEY (k, v0))",
+                    consistency: :all)
+
+    # Insert wide rows
+    insert = session.prepare("INSERT INTO testreadfail.test2 (k, v0, v1) VALUES (1, ?, 1)")
+    (0..3000).each do |num|
+      session.execute(insert, arguments: [num])
+    end
 
-      # Insert wide rows
-      insert = session.prepare("INSERT INTO testreadfail.test2 (k, v0, v1) VALUES (1, ?, 1)")
-      (0..3000).each do |num|
-        session.execute(insert, arguments: [num])
-      end
+    # Delete wide rows
+    delete = session.prepare("DELETE v1 FROM testreadfail.test2 WHERE k = 1 AND v0 =?")
+    (0..2001).each do |num|
+      session.execute(delete, arguments: [num])
+    end
 
-      # Delete wide rows
-      delete = session.prepare("DELETE v1 FROM testreadfail.test2 WHERE k = 1 AND v0 =?")
-      (0..2001).each do |num|
-        session.execute(delete, arguments: [num])
-      end
+    # Tombstones should trigger ReadFailure
+    ex = assert_raises(Cassandra::Errors::ReadError) do
+      session.execute("SELECT * FROM testreadfail.test2 WHERE k = 1")
+    end
 
-      # Tombstones should trigger ReadFailure
-      assert_raises(Cassandra::Errors::ReadError) do
-        session.execute("SELECT * FROM testreadfail.test2 WHERE k = 1")
-      end
-    ensure
-      cluster && cluster.close
+    # If we're speaking protocol version > 4, verify that we have a failure map in the exception.
+    connection_options = cluster.instance_variable_get(:@connection_options)
+    assert_operator(ex.failed, :>=, 1)
+    if connection_options.protocol_version > 4
+      assert_equal(ex.failed, ex.failures_by_node.size)
+      assert_equal(1, ex.failures_by_node.values.first)
     end
+  ensure
+    cluster && cluster.close
   end
 
   # Test for validating FunctionCallError
@@ -165,30 +188,28 @@ def test_raise_error_on_read_failure
   def test_raise_error_on_function_failure
     skip("Client failure errors are only available in C* after 2.2") if CCM.cassandra_version < '2.2.0'
 
-    begin
-      cluster = Retry.with_attempts(5, Cassandra::Errors::NoHostsAvailable) { Cassandra.cluster }
-      session = cluster.connect
+    cluster = Retry.with_attempts(5, Cassandra::Errors::NoHostsAvailable) { Cassandra.cluster }
+    session = cluster.connect
 
-      session.execute("CREATE KEYSPACE testfunctionfail WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}",
-                      consistency: :all) rescue nil
-      session.execute("CREATE TABLE testfunctionfail.d (k int PRIMARY KEY , d double)", consistency: :all)
+    session.execute("CREATE KEYSPACE testfunctionfail WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '3'}",
+                    consistency: :all) rescue nil
+    session.execute("CREATE TABLE IF NOT EXISTS testfunctionfail.d (k int PRIMARY KEY , d double)", consistency: :all)
 
-      # Create a UDF that throws an exception
-      session.execute("CREATE FUNCTION testfunctionfail.test_failure(d double)
+    # Create a UDF that throws an exception
+    session.execute("CREATE FUNCTION IF NOT EXISTS testfunctionfail.test_failure(d double)
                       RETURNS NULL ON NULL INPUT
                       RETURNS double
                       LANGUAGE java AS 'throw new RuntimeException(\"failure\");'",
-                      consistency: :all)
+                    consistency: :all)
 
-      # Insert value to use for function
-      session.execute("INSERT INTO testfunctionfail.d (k, d) VALUES (0, 5.12)")
+    # Insert value to use for function
+    session.execute("INSERT INTO testfunctionfail.d (k, d) VALUES (0, 5.12)")
 
-      # FunctionFailure should be triggered
-      assert_raises(Cassandra::Errors::FunctionCallError) do
-        session.execute("SELECT test_failure(d) FROM testfunctionfail.d WHERE k = 0")
-      end
-    ensure
-      cluster && cluster.close
+    # FunctionFailure should be triggered
+    assert_raises(Cassandra::Errors::FunctionCallError) do
+      session.execute("SELECT test_failure(d) FROM testfunctionfail.d WHERE k = 0")
     end
+  ensure
+    cluster && cluster.close
   end
 end

lib/cassandra.rb

@@ -56,6 +56,7 @@ module Cassandra
   CLUSTER_OPTIONS = [
     :address_resolution,
     :address_resolution_policy,
+    :allow_beta_protocol,
     :auth_provider,
     :client_cert,
     :client_timestamps,
@@ -196,6 +197,9 @@ module Cassandra
   #   nodes. By default, this is auto-negotiated to the highest common protocol version
   #   that all nodes in `:hosts` speak.
   #
+  # @option options [Boolean] :allow_beta_protocol (false) whether the driver should attempt to speak to nodes
+  #   with a beta version of the newest protocol (which is still under development). USE WITH CAUTION!
+  #
   # @option options [Boolean, Cassandra::TimestampGenerator] :client_timestamps (false) whether the driver
   #   should send timestamps for each executed statement and possibly which timestamp generator to use. Enabling this
   #   setting helps mitigate Cassandra cluster clock skew because the timestamp of the client machine will be used.
@@ -658,6 +662,7 @@ def self.validate_and_massage_options(options)
     options[:nodelay] = !!options[:nodelay] if options.key?(:nodelay)
     options[:trace] = !!options[:trace] if options.key?(:trace)
     options[:shuffle_replicas] = !!options[:shuffle_replicas] if options.key?(:shuffle_replicas)
+    options[:allow_beta_protocol] = !!options[:allow_beta_protocol] if options.key?(:allow_beta_protocol)
 
     if options.key?(:page_size)
       page_size = options[:page_size]
@@ -675,12 +680,16 @@ def self.validate_and_massage_options(options)
       protocol_version = options[:protocol_version]
       unless protocol_version.nil?
         Util.assert_instance_of(::Integer, protocol_version)
-        Util.assert_one_of(1..4, protocol_version) do
-          ":protocol_version must be a positive integer, #{protocol_version.inspect} given"
-        end
+        Util.assert_one_of(1..Cassandra::Protocol::Versions::MAX_SUPPORTED_VERSION, protocol_version,
+          ':protocol_version must be a positive integer between 1 and ' \
+          "#{Cassandra::Protocol::Versions::MAX_SUPPORTED_VERSION}, #{protocol_version.inspect} given"
+        )
       end
     end
 
+    Util.assert(!(options[:allow_beta_protocol] && options[:protocol_version]),
+                'only one of :allow_beta_protocol and :protocol_version may be specified, both given')
+
     if options.key?(:futures_factory)
       futures_factory = options[:futures_factory]
       methods = [:error, :value, :promise, :all]

lib/cassandra/cluster/connector.rb

@@ -122,11 +122,8 @@ def do_connect(host)
                          ssl: @connection_options.ssl) do |connection|
           raise Errors::ClientError, 'Not connected, reactor stopped' unless connection
 
-          if @connection_options.nodelay?
-            connection.to_io.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)
-          else
-            connection.to_io.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 0)
-          end
+          connection.to_io.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY,
+                                      @connection_options.nodelay? ? 1 : 0)
 
           Protocol::CqlProtocolHandler.new(connection,
                                            @reactor,

lib/cassandra/cluster/options.rb

@@ -25,7 +25,7 @@ class Options
       attr_reader :auth_provider, :compressor, :connect_timeout, :credentials,
                   :heartbeat_interval, :idle_timeout, :port, :schema_refresh_delay,
                   :schema_refresh_timeout, :ssl, :custom_type_handlers
-      attr_boolean :protocol_negotiable, :synchronize_schema, :nodelay
+      attr_boolean :protocol_negotiable, :synchronize_schema, :nodelay, :allow_beta_protocol
 
       attr_accessor :protocol_version
 
@@ -46,7 +46,8 @@ def initialize(logger,
                      schema_refresh_timeout,
                      nodelay,
                      requests_per_connection,
-                     custom_types)
+                     custom_types,
+                     allow_beta_protocol)
         @logger                 = logger
         @protocol_version       = protocol_version
         @credentials            = credentials
@@ -61,6 +62,7 @@ def initialize(logger,
         @schema_refresh_delay   = schema_refresh_delay
         @schema_refresh_timeout = schema_refresh_timeout
         @nodelay                = nodelay
+        @allow_beta_protocol    = allow_beta_protocol
         @custom_type_handlers   = {}
         custom_types.each do |type_klass|
           @custom_type_handlers[type_klass.type] = type_klass
@@ -71,12 +73,14 @@ def initialize(logger,
         @requests_per_connection = requests_per_connection
 
         # If @protocol_version is nil, it means we want the driver to negotiate the
-        # protocol starting with our known max (4). If @protocol_version is not nil,
+        # protocol starting with our known max. If @protocol_version is not nil,
         # it means the user wants us to use a particular version, so we should not
         # support negotiation.
 
         @protocol_negotiable = @protocol_version.nil?
-        @protocol_version ||= 4
+        @protocol_version ||= allow_beta_protocol ?
+            Cassandra::Protocol::Versions::BETA_VERSION :
+            Cassandra::Protocol::Versions::MAX_SUPPORTED_VERSION
       end
 
       def compression

lib/cassandra/driver.rb

@@ -141,13 +141,15 @@ def self.let(name, &block)
         schema_refresh_timeout,
         nodelay,
         requests_per_connection,
-        custom_types
+        custom_types,
+        allow_beta_protocol
       )
     end
 
     let(:custom_types)      { [] }
     let(:port)                      { 9042 }
     let(:protocol_version)          { nil }
+    let(:allow_beta_protocol)       { false }
     let(:connect_timeout)           { 10 }
     let(:ssl)                       { false }
     let(:logger)                    { NullLogger.new }

lib/cassandra/errors.rb

@@ -348,6 +348,9 @@ class WriteError < ::StandardError
       attr_reader :received
       # @return [Integer] the number of writes failed
       attr_reader :failed
+      # @return [Hash<IPAddr, Integer>] map of <ip, error-code>. This is new in v5 and is nil in previous versions
+      #    of the Casssandra protocol.
+      attr_reader :failures_by_node
 
       # @private
       def initialize(message,
@@ -363,7 +366,8 @@ def initialize(message,
                      consistency,
                      required,
                      failed,
-                     received)
+                     received,
+                     failures_by_node)
         super(message,
               payload,
               warnings,
@@ -378,6 +382,7 @@ def initialize(message,
         @required    = required
         @failed      = failed
         @received    = received
+        @failures_by_node = failures_by_node
       end
     end
 
@@ -400,6 +405,9 @@ class ReadError < ::StandardError
       attr_reader :received
       # @return [Integer] the number of reads failed
       attr_reader :failed
+      # @return [Hash<IPaddr, Integer>] map of <ip, error-code>. This is new in v5 and is nil in previous versions
+      #    of the Casssandra protocol.
+      attr_reader :failures_by_node
 
       # @private
       def initialize(message,
@@ -415,7 +423,8 @@ def initialize(message,
                      consistency,
                      required,
                      failed,
-                     received)
+                     received,
+                     failures_by_node)
         super(message,
               payload,
               warnings,
@@ -430,6 +439,7 @@ def initialize(message,
         @required    = required
         @failed      = failed
         @received    = received
+        @failures_by_node = failures_by_node
       end
 
       def retrieved?

lib/cassandra/protocol.rb

@@ -49,6 +49,11 @@ module Constants
       SCHEMA_CHANGE_TARGET_FUNCTION  = 'FUNCTION'.freeze
       SCHEMA_CHANGE_TARGET_AGGREGATE = 'AGGREGATE'.freeze
     end
+
+    module Versions
+      BETA_VERSION = 5
+      MAX_SUPPORTED_VERSION = 4
+    end
   end
 end