Merge branch 'master' of https://github.com/datastax/python-driver into sync_with_upstream

* 'master' of https://github.com/datastax/python-driver:
  Merge pull request datastax#1126 from eamanu/fix-typos
  PYTHON-1294: Upgrade importlib-metadata to a much newer version
  Add tests for recent addition of execution profile support to cassandra.concurrent
  Merge pull request datastax#1122 from andy-slac/concurrent-execution-profiles
  Merge pull request datastax#1119 from datastax/python-1290
  Merge pull request datastax#1117 from datastax/remove_unittest2
  Removing file unexpectedly included in previous PR
  Merge pull request datastax#1114 from haaawk/stream_ids_fix
  Merge pull request datastax#1116 from Orenef11/fix_default_argument_value
  Comment update following off of datastax#1110
  Merge pull request datastax#1103 from numberly/fix_empty_paging
  Merge pull request datastax#1103 from psarna/fix_deprecation_in_tracing
  Fixes to the Travis build. (datastax#1111)

Author: fruch

cassandra/cluster.py

@@ -2451,7 +2451,7 @@ def default_consistency_level(self, cl):
         *Deprecated:* use execution profiles instead
         """
         warn("Setting the consistency level at the session level will be removed in 4.0. Consider using "
-             "execution profiles and setting the desired consitency level to the EXEC_PROFILE_DEFAULT profile."
+             "execution profiles and setting the desired consistency level to the EXEC_PROFILE_DEFAULT profile."
              , DeprecationWarning)
         self._validate_set_legacy_config('default_consistency_level', cl)
 
@@ -5212,6 +5212,12 @@ def next(self):
             self._page_iter = iter(self._current_rows)
             return self.next()
 
+            # Some servers can return empty pages in this case; Scylla is known to do
+            # so in some circumstances.  Guard against this by recursing to handle
+            # the next(iter) call.  If we have an empty page in that case it will
+            # get handled by the StopIteration handler when we recurse.
+            return self.next()
+
         return next(self._page_iter)
 
     __next__ = next

cassandra/concurrent.py

@@ -21,15 +21,15 @@
 from threading import Condition
 import sys
 
-from cassandra.cluster import ResultSet
+from cassandra.cluster import ResultSet, EXEC_PROFILE_DEFAULT
 
 import logging
 log = logging.getLogger(__name__)
 
 
 ExecutionResult = namedtuple('ExecutionResult', ['success', 'result_or_exc'])
 
-def execute_concurrent(session, statements_and_parameters, concurrency=100, raise_on_first_error=True, results_generator=False):
+def execute_concurrent(session, statements_and_parameters, concurrency=100, raise_on_first_error=True, results_generator=False, execution_profile=EXEC_PROFILE_DEFAULT):
     """
     Executes a sequence of (statement, parameters) tuples concurrently.  Each
     ``parameters`` item must be a sequence or :const:`None`.
@@ -56,6 +56,9 @@ def execute_concurrent(session, statements_and_parameters, concurrency=100, rais
       footprint is marginal CPU overhead (more thread coordination and sorting out-of-order results
       on-the-fly).
 
+    `execution_profile` argument is the execution profile to use for this
+    request, it is passed directly to :meth:`Session.execute_async`.
+
     A sequence of ``ExecutionResult(success, result_or_exc)`` namedtuples is returned
     in the same order that the statements were passed in.  If ``success`` is :const:`False`,
     there was an error executing the statement, and ``result_or_exc`` will be
@@ -90,17 +93,19 @@ def execute_concurrent(session, statements_and_parameters, concurrency=100, rais
     if not statements_and_parameters:
         return []
 
-    executor = ConcurrentExecutorGenResults(session, statements_and_parameters) if results_generator else ConcurrentExecutorListResults(session, statements_and_parameters)
+    executor = ConcurrentExecutorGenResults(session, statements_and_parameters, execution_profile) \
+        if results_generator else ConcurrentExecutorListResults(session, statements_and_parameters, execution_profile)
     return executor.execute(concurrency, raise_on_first_error)
 
 
 class _ConcurrentExecutor(object):
 
     max_error_recursion = 100
 
-    def __init__(self, session, statements_and_params):
+    def __init__(self, session, statements_and_params, execution_profile):
         self.session = session
         self._enum_statements = enumerate(iter(statements_and_params))
+        self._execution_profile = execution_profile
         self._condition = Condition()
         self._fail_fast = False
         self._results_queue = []
@@ -132,7 +137,7 @@ def _execute_next(self):
     def _execute(self, idx, statement, params):
         self._exec_depth += 1
         try:
-            future = self.session.execute_async(statement, params, timeout=None)
+            future = self.session.execute_async(statement, params, timeout=None, execution_profile=self._execution_profile)
             args = (future, idx)
             future.add_callbacks(
                 callback=self._on_success, callback_args=args,

cassandra/connection.py

@@ -769,6 +769,8 @@ class Connection(object):
 
     _is_checksumming_enabled = False
 
+    _on_orphaned_stream_released = None
+
     @property
     def _iobuf(self):
         # backward compatibility, to avoid any change in the reactors
@@ -778,8 +780,8 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
                  ssl_options=None, sockopts=None, compression=True,
                  cql_version=None, protocol_version=ProtocolVersion.MAX_SUPPORTED, is_control_connection=False,
                  user_type_map=None, connect_timeout=None, allow_beta_protocol_version=False, no_compact=False,
-                 ssl_context=None, owning_pool=None, shard_id=None, total_shards=None):
-
+                 ssl_context=None, owning_pool=None, shard_id=None, total_shards=None,
+                 on_orphaned_stream_released=None):
         # TODO next major rename host to endpoint and remove port kwarg.
         self.endpoint = host if isinstance(host, EndPoint) else DefaultEndPoint(host, port)
 
@@ -801,7 +803,7 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
         self._continuous_paging_sessions = {}
         self._socket_writable = True
         self.orphaned_request_ids = set()
-        self._owning_pool = owning_pool
+        self._on_orphaned_stream_released = on_orphaned_stream_released
 
         if ssl_options:
             self._check_hostname = bool(self.ssl_options.pop('check_hostname', False))
@@ -1247,8 +1249,8 @@ def process_msg(self, header, body):
                         self.in_flight -= 1
                         self.orphaned_request_ids.remove(stream_id)
                         need_notify_of_release = True
-                if need_notify_of_release and self._owning_pool:
-                    self._owning_pool.on_orphaned_stream_released()
+                if need_notify_of_release and self._on_orphaned_stream_released:
+                    self._on_orphaned_stream_released()
 
                 try:
                     callback, decoder, result_metadata = self._requests.pop(stream_id)

cassandra/io/asyncioreactor.py

@@ -46,9 +46,8 @@ def __init__(self, timeout, callback, loop):
         self._handle = asyncio.run_coroutine_threadsafe(delayed, loop=loop)
 
     @staticmethod
-    @asyncio.coroutine
-    def _call_delayed_coro(timeout, callback, loop):
-        yield from asyncio.sleep(timeout, loop=loop)
+    async def _call_delayed_coro(timeout, callback, loop):
+        await asyncio.sleep(timeout, loop=loop)
         return callback()
 
     def __lt__(self, other):
@@ -136,8 +135,7 @@ def close(self):
             self._close(), loop=self._loop
         )
 
-    @asyncio.coroutine
-    def _close(self):
+    async def _close(self):
         log.debug("Closing connection (%s) to %s" % (id(self), self.endpoint))
         if self._write_watcher:
             self._write_watcher.cancel()
@@ -174,40 +172,39 @@ def push(self, data):
             # avoid races/hangs by just scheduling this, not using threadsafe
             self._loop.create_task(self._push_msg(chunks))
 
-    @asyncio.coroutine
-    def _push_msg(self, chunks):
+    async def _push_msg(self, chunks):
         # This lock ensures all chunks of a message are sequential in the Queue
-        with (yield from self._write_queue_lock):
+        with await self._write_queue_lock:
             for chunk in chunks:
                 self._write_queue.put_nowait(chunk)
 
 
-    @asyncio.coroutine
-    def handle_write(self):
+    async def handle_write(self):
         while True:
             try:
-                next_msg = yield from self._write_queue.get()
+                next_msg = await self._write_queue.get()
                 if next_msg:
-                    yield from self._loop.sock_sendall(self._socket, next_msg)
+                    await self._loop.sock_sendall(self._socket, next_msg)
             except socket.error as err:
                 log.debug("Exception in send for %s: %s", self, err)
                 self.defunct(err)
                 return
             except asyncio.CancelledError:
                 return
 
-    @asyncio.coroutine
-    def handle_read(self):
+    async def handle_read(self):
         while True:
             try:
-                buf = yield from self._loop.sock_recv(self._socket, self.in_buffer_size)
+                buf = await self._loop.sock_recv(self._socket, self.in_buffer_size)
                 self._iobuf.write(buf)
             # sock_recv expects EWOULDBLOCK if socket provides no data, but
             # nonblocking ssl sockets raise these instead, so we handle them
             # ourselves by yielding to the event loop, where the socket will
             # get the reading/writing it "wants" before retrying
             except (ssl.SSLWantWriteError, ssl.SSLWantReadError):
-                yield
+                # Apparently the preferred way to yield to the event loop from within
+                # a native coroutine based on https://github.com/python/asyncio/issues/284
+                await asyncio.sleep(0)
                 continue
             except socket.error as err:
                 log.debug("Exception during socket recv for %s: %s",

cassandra/pool.py

@@ -423,7 +423,7 @@ def __init__(self, host, host_distance, session):
             return
 
         log.debug("Initializing connection for host %s", self.host)
-        first_connection = session.cluster.connection_factory(self.host.endpoint, owning_pool=self)
+        first_connection = session.cluster.connection_factory(self.host.endpoint, on_orphaned_stream_released=self.on_orphaned_stream_released)
         log.debug("First connection created to %s for shard_id=%i", self.host, first_connection.shard_id)
         self._connections[first_connection.shard_id] = first_connection
         self._keyspace = session.keyspace
@@ -517,7 +517,10 @@ def borrow_connection(self, timeout, routing_key=None):
                     last_retry = True
                     continue
             with self._stream_available_condition:
-                self._stream_available_condition.wait(remaining)
+                if conn.orphaned_threshold_reached and conn.is_closed:
+                    conn = self._get_connection()
+                else:
+                    self._stream_available_condition.wait(remaining)
 
         raise NoConnectionsAvailable("All request IDs are currently in use")
 
@@ -563,8 +566,8 @@ def return_connection(self, connection, stream_was_orphaned=False):
                         with self._lock:
                             if connection in self._trash:
                                 self._trash.remove(connection)
-                        log.debug("Closing trashed connection (%s) to %s", id(connection), self.host)
-                        connection.close()
+                                log.debug("Closing trashed connection (%s) to %s", id(connection), self.host)
+                                connection.close()
                 return
 
     def on_orphaned_stream_released(self):
@@ -588,7 +591,8 @@ def _replace(self, connection):
                     self._connecting.add(connection.shard_id)
                     self._session.submit(self._open_connection_to_missing_shard, connection.shard_id)
                 else:
-                    connection = self._session.cluster.connection_factory(self.host.endpoint, owning_pool=self)
+                    connection = self._session.cluster.connection_factory(self.host.endpoint,
+                                                                          on_orphaned_stream_released=self.on_orphaned_stream_released)
                     if self._keyspace:
                         connection.set_keyspace_blocking(self._keyspace)
                     self._connections[connection.shard_id] = connection
@@ -690,12 +694,12 @@ def _open_connection_to_missing_shard(self, shard_id):
         log.debug("shard_aware_endpoint=%r", shard_aware_endpoint)
 
         if shard_aware_endpoint:
-            conn = self._session.cluster.connection_factory(shard_aware_endpoint, owning_pool=self,
+            conn = self._session.cluster.connection_factory(shard_aware_endpoint, on_orphaned_stream_released=self.on_orphaned_stream_released,
                                                             shard_id=shard_id,
                                                             total_shards=self.host.sharding_info.shards_count)
             conn.original_endpoint = self.host.endpoint
         else:
-            conn = self._session.cluster.connection_factory(self.host.endpoint, owning_pool=self)
+            conn = self._session.cluster.connection_factory(self.host.endpoint, on_orphaned_stream_released=self.on_orphaned_stream_released)
 
         log.debug("Received a connection %s for shard_id=%i on host %s", id(conn), conn.shard_id, self.host)
         if self.is_shutdown:
@@ -827,6 +831,16 @@ def _open_connections_for_all_shards(self, skip_shard_id=None):
                     self._connecting.add(shard_id)
                     self._shard_connections_futures.append(future)
 
+        trash_conns = None
+        with self._lock:
+            if self._trash:
+                trash_conns = self._trash
+                self._trash = set()
+
+        if trash_conns is not None:
+            for conn in self._trash:
+                conn.close()
+
     def _set_keyspace_for_all_conns(self, keyspace, callback):
         """
         Asynchronously sets the keyspace for all connections.  When all
@@ -905,7 +919,7 @@ def __init__(self, host, host_distance, session):
 
         log.debug("Initializing new connection pool for host %s", self.host)
         core_conns = session.cluster.get_core_connections_per_host(host_distance)
-        self._connections = [session.cluster.connection_factory(host.endpoint, owning_pool=self)
+        self._connections = [session.cluster.connection_factory(host.endpoint, on_orphaned_stream_released=self.on_orphaned_stream_released)
                              for i in range(core_conns)]
 
         self._keyspace = session.keyspace
@@ -1009,7 +1023,7 @@ def _add_conn_if_under_max(self):
 
         log.debug("Going to open new connection to host %s", self.host)
         try:
-            conn = self._session.cluster.connection_factory(self.host.endpoint, owning_pool=self)
+            conn = self._session.cluster.connection_factory(self.host.endpoint, on_orphaned_stream_released=self.on_orphaned_stream_released)
             if self._keyspace:
                 conn.set_keyspace_blocking(self._session.keyspace)
             self._next_trash_allowed_at = time.time() + _MIN_TRASH_INTERVAL

cassandra/query.py

@@ -996,7 +996,8 @@ def populate(self, max_wait=2.0, wait_for_complete=True, query_cl=None):
                 SimpleStatement(self._SELECT_SESSIONS_FORMAT, consistency_level=query_cl), (self.trace_id,), time_spent, max_wait)
 
             # PYTHON-730: There is race condition that the duration mutation is written before started_at the for fast queries
-            is_complete = session_results and session_results[0].duration is not None and session_results[0].started_at is not None
+            session_row = session_results.one() if session_results else None
+            is_complete = session_row is not None and session_row.duration is not None and session_row.started_at is not None
             if not session_results or (wait_for_complete and not is_complete):
                 time.sleep(self._BASE_RETRY_SLEEP * (2 ** attempt))
                 attempt += 1
@@ -1006,7 +1007,6 @@ def populate(self, max_wait=2.0, wait_for_complete=True, query_cl=None):
             else:
                 log.debug("Fetching parital trace info for trace ID: %s", self.trace_id)
 
-            session_row = session_results[0]
             self.request_type = session_row.request
             self.duration = timedelta(microseconds=session_row.duration) if is_complete else None
             self.started_at = session_row.started_at

cassandra/util.py

@@ -797,7 +797,7 @@ class OrderedMap(Mapping):
     '''
     An ordered map that accepts non-hashable types for keys. It also maintains the
     insertion order of items, behaving as OrderedDict in that regard. These maps
-    are constructed and read just as normal mapping types, exept that they may
+    are constructed and read just as normal mapping types, except that they may
     contain arbitrary collections and other non-hashable items as keys::
 
         >>> od = OrderedMap([({'one': 1, 'two': 2}, 'value'),

test-requirements.txt

@@ -2,8 +2,6 @@
 scales
 nose
 mock>1.1
-#ccm>=2.1.2
-unittest2; python_version < '3.5'
 pytz
 sure
 pure-sasl

tests/__init__.py

@@ -12,10 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-try:
-    import unittest2 as unittest
-except ImportError:
-    import unittest  # noqa
+import unittest
 import logging
 import sys
 import socket

tests/integration/__init__.py

@@ -20,10 +20,7 @@
 if connection_class is not None:
     Cluster.connection_class = connection_class
 
-try:
-    import unittest2 as unittest
-except ImportError:
-    import unittest  # noqa
+import unittest
 
 from packaging.version import Version
 import logging
@@ -405,15 +402,15 @@ def get_node(node_id):
     return CCM_CLUSTER.nodes['node%s' % node_id]
 
 
-def use_multidc(dc_list, workloads=[]):
+def use_multidc(dc_list, workloads=None):
     use_cluster(MULTIDC_CLUSTER_NAME, dc_list, start=True, workloads=workloads)
 
 
-def use_singledc(start=True, workloads=[], use_single_interface=USE_SINGLE_INTERFACE):
+def use_singledc(start=True, workloads=None, use_single_interface=USE_SINGLE_INTERFACE):
     use_cluster(CLUSTER_NAME, [3], start=start, workloads=workloads, use_single_interface=use_single_interface)
 
 
-def use_single_node(start=True, workloads=[], configuration_options={}, dse_options={}):
+def use_single_node(start=True, workloads=None, configuration_options=None, dse_options=None):
     use_cluster(SINGLE_NODE_CLUSTER_NAME, [1], start=start, workloads=workloads,
                 configuration_options=configuration_options, dse_options=dse_options)
 
@@ -475,10 +472,11 @@ def start_cluster_wait_for_up(cluster):
 
 
 def use_cluster(cluster_name, nodes, ipformat=None, start=True, workloads=None, set_keyspace=True, ccm_options=None,
-                configuration_options={}, dse_options={}, use_single_interface=USE_SINGLE_INTERFACE):
+                configuration_options=None, dse_options=None, use_single_interface=USE_SINGLE_INTERFACE):
+    configuration_options = configuration_options or {}
+    dse_options = dse_options or {}
+    workloads = workloads or []
     dse_cluster = True if DSE_VERSION else False
-    if not workloads:
-        workloads = []
 
     if ccm_options is None and DSE_VERSION:
         ccm_options = {"version": CCM_VERSION}

tests/integration/advanced/__init__.py

@@ -12,10 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-try:
-    import unittest2 as unittest
-except ImportError:
-    import unittest  # noqa
+import unittest
 
 from six.moves.urllib.request import build_opener, Request, HTTPHandler
 import re