add cassandra_options (#4224)

* add cassandra_options

* fix misspelled `cassandra_options`

* add cassandra_options to defaults

* cassandra options match auth_kwargs

celery/app/defaults.py

@@ -126,6 +126,7 @@ NAMESPACES = Namespace(
         write_consistency=Option(type='string'),
         auth_provider=Option(type='string'),
         auth_kwargs=Option(type='string'),
+        options=Option({}, type='dict'),
     ),
     control=Namespace(
         queue_ttl=Option(300.0, type='float'),

celery/backends/cassandra.py

@@ -90,6 +90,7 @@ class CassandraBackend(BaseBackend):
         self.port = port or conf.get('cassandra_port', None)
         self.keyspace = keyspace or conf.get('cassandra_keyspace', None)
         self.table = table or conf.get('cassandra_table', None)
+        self.cassandra_options = conf.get('cassandra_options', {})
 
         if not self.servers or not self.keyspace or not self.table:
             raise ImproperlyConfigured('Cassandra backend not configured.')
@@ -141,7 +142,8 @@ class CassandraBackend(BaseBackend):
         try:
             self._connection = cassandra.cluster.Cluster(
                 self.servers, port=self.port,
-                auth_provider=self.auth_provider)
+                auth_provider=self.auth_provider,
+                **self.cassandra_options)
             self._session = self._connection.connect(self.keyspace)
 
             # We're forced to do concatenation below, as formatting would

docs/userguide/configuration.rst

@@ -83,6 +83,7 @@ rush in moving to the new settings format.
 ``CASSANDRA_READ_CONSISTENCY``         :setting:`cassandra_read_consistency`
 ``CASSANDRA_SERVERS``                  :setting:`cassandra_servers`
 ``CASSANDRA_WRITE_CONSISTENCY``        :setting:`cassandra_write_consistency`
+``CASSANDRA_OPTIONS``                  :setting:`cassandra_options`
 ``CELERY_COUCHBASE_BACKEND_SETTINGS``  :setting:`couchbase_backend_settings`
 ``CELERY_MONGODB_BACKEND_SETTINGS``    :setting:`mongodb_backend_settings`
 ``CELERY_EVENT_QUEUE_EXPIRES``         :setting:`event_queue_expires`
@@ -1038,6 +1039,22 @@ Named arguments to pass into the authentication provider. For example:
         password: 'cassandra'
     }
 
+.. setting:: cassandra_options
+
+``cassandra_options``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Default: ``{}`` (empty mapping).
+
+Named arguments to pass into the ``cassandra.cluster`` class.
+
+.. code-block:: python
+
+    cassandra_options = {
+        'cql_version': '3.2.1'
+        'protocol_version': 3
+    }
+
 Example configuration
 ~~~~~~~~~~~~~~~~~~~~~
 

t/unit/backends/test_cassandra.py

@@ -182,3 +182,15 @@ class test_CassandraBackend:
         }
         with pytest.raises(ImproperlyConfigured):
             mod.CassandraBackend(app=self.app)
+
+    def test_options(self):
+        # Ensure valid options works properly
+        from celery.backends import cassandra as mod
+
+        mod.cassandra = Mock()
+        # Valid options
+        self.app.conf.cassandra_options = {
+            'cql_version': '3.2.1',
+            'protocol_version': 3
+        }
+        mod.CassandraBackend(app=self.app)